代理型人工智能（AI）有望成为2025年最大的人工智能趋势之一。得益于科技公司的大量投资，消费者能够以更低的成本获得功能更强大、上下文窗口不断扩大的模型。到2025年，我们将开始看到人工智能更高级的应用案例，超越如检索增强系统（RAG）等复合AI系统。

毫无疑问，我们正处于一个人工智能以惊人速度加速发展的时代。我们无需回望太远，就能见证过去几年中发布的机器学习工具、平台和框架数量的迅速增长。

人工智能正变得越来越强大，这在一定程度上要归功于代理框架，它使人工智能能够执行更加复杂的任务。代理工具集成已成为扩展人工智能能力的重要组成部分，赋予了人工智能感知和与世界互动的能力。

终于有可能开发出能够像我们的个人礼宾服务一样自主行动的智能体，它们可以为我们预订旅行，或者作为我们的私人导师帮助我们准备会议，或者在代码配对会议中扮演高级开发者的角色。代理型人工智能所带来的经济价值可能是巨大的。

代理工具推动了我们所看到的模型性能提升。已经证明，与仅使用提示或甚至微调相比，工具使用显著提升了模型的性能。

因此，在我看来，Anthropic于2024年11月开源的模型上下文协议（Model Context Protocol，简称MCP）可能是大语言模型（LLM）领域最具变革性的事件之一。以下是原因。

什么是代理型人工智能？

代理型人工智能定义了一种能够自主行动以完成任务的人工智能类型。最重要的是，它是目标导向的，并且在解决问题的方式上具有多样性。代理型人工智能解决问题的方式更接近我们人类完成任务的方式。

与遵循程序定义的控制逻辑的复合人工智能系统（类似于数据工程管道）不同，代理型人工智能所采取的步骤定义更为宽松。然而，代理型人工智能系统的核心仍然是一个大语言模型，但其角色得到了增强，并被定位为整个代理系统的推理中心。

在解决问题的每个阶段，代理型人工智能都会采用一个四步过程：

1. 感知：人工智能智能体会从各种数据源（如数据库或外部API调用）构建上下文基础。
2. 推理：根据上下文信息，大语言模型作为推理引擎来理解任务，识别其理解中的缺口，以及实现目标所需的任何进一步行动。
3. 行动：代理型人工智能会执行计划，这可能包括调用已集成到其系统中的任何外部工具。工具可能包括用于执行git提交的Github API调用，或使智能体能够与你的电子邮件收件箱交互的工具。
4. 学习：代理型人工智能可以从之前的交互中学习和适应，以持续改进。

代理型人工智能系统具有高度适应性，非常适合上下文环境不断变化的动态问题。更重要的是，这种适应性以智能体感知和与世界互动的能力为基础。这就是模型上下文协议可以发挥作用的地方。

模型上下文协议

模型上下文协议（MCP）是Anthropic发布的一个开源标准。模型上下文协议强调了通过开放协议进行工具标准化的重要性。具体来说，它标准化了应用程序如何与大语言模型交互，并为大语言模型提供上下文。就像HTTP标准化了我们如何在互联网上通信一样，MCP为大语言模型与外部工具交互提供了一个标准协议。

代理型人工智能中使用的解决问题过程往往需要大语言模型与数据和工具集成。MCP协议提出了一种结构化方法，定义了应用程序如何为大语言模型提供上下文。MCP遵循客户端-服务器架构，其中主机应用程序可以连接到多个服务器。服务器与主机之间的通信是双向的，每个服务器都连接到一个数据源或远程服务，大语言模型可以与之集成。双向通信使数据源能够与大语言模型共享其结构、内容和可用工具。

MCP是代理型应用中更复杂工作流的推动者。更重要的是，MCP使用户能够共享预构建的集成，这些集成可以直接导入到现有项目中，而无需进行重大的配置更改。

让我们来看看如何创建自己的MCP服务并将其与LangChain集成。

内容

1. 前提条件
2. 创建自己的MCP服务器
3. 测试你的MCP服务器
4. 定义环境变量
5. 将MCP服务器连接到LangChain

前提条件

在我们开始之前，你需要满足几个前提条件。你需要克隆这里链接的github项目。该项目利用了我编写的一个开源项目langchain-mcp-connect，该项目在github和pypi上都有链接。

导航到项目的根目录并使用uv安装 python 依赖项：

uv syncsync

激活Python虚拟环境，使用：

source .venv/bin/activate

你需要在根目录下创建一个 .env 文件，并包含以下值：

OPENAI_API_KEY=<YOUR_API_TOKEN_HERE>
GITHUB_PERSONAL_ACCESS_TOKEN=<YOUR_PERSONAL_ACCESS_TOKEN_HERE>

或者，你可以直接在当前的 shell 中定义 GITHUB_PERSONAL_ACCESS_TOKEN，方法如下：

export OPENAI_API_KEY=<YOUR_API_TOKEN_HERE>
export GITHUB_PERSONAL_ACCESS_TOKEN=<YOUR_PERSONAL_ACCESS_TOKEN_HERE>

创建你自己的 MCP 服务器

模型上下文协议（MCP）提供了一种标准化的方式来构建可与任何大型语言模型（LLM）集成的自定义服务器。值得注意的是，MCP 在其架构中全程使用 pydantic，以确保系统所有组件之间的强大数据验证和类型安全。MCP 提供了有用的装饰器，以简化服务器 API 的创建。你的服务器运行需要两个方法。

list_tool — 此方法返回 LLM 可调用的工具列表。最重要的是，此方法详细说明了名称、描述以及为 LLM 设置上下文的输入模式。下面的方法描述了两个用于调用 GitHub API 的方法，一个是 get_file_contents 方法，另一个是 list_repo_tree 方法。

@app.list_tools()
async def list_tools() -> list[Tool]:
    """List available weather tools."""
    return [
        Tool(
            name="get_file_contents",
            description="Retrieve the file contents of a file from a github repository",
            inputSchema=GetFileContentsArgs.model_json_schema(),
        ),
        Tool(
            name="list_repo_tree",
            description="List the tree structure of a GitHub repository",
            inputSchema=GetTreeArgs.model_json_schema(),
        ),
    ]

call_tool — 当调用 MCP 服务器时，会启动此方法。此方法接受两个参数，第一个是方法的名称，第二个是方法的参数。简而言之，call_tool 方法将流量路由到正确的函数调用。下面的方法描述了两条可能的路由，即 get_file_contents 方法和 list_repo_tree 方法。

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle the call_tool for the vector store service."""
    match name:
        case GitHubTools.GET_FILE_CONTENTS:
            args = GetFileContentsArgs(**arguments)
            file = get_file_contents(args)
            return [TextContent(type="text", text=file.content)]
        case GitHubTools.LIST_REPO_TREE:
            args = GetTreeArgs(**arguments)
            tree = list_repo_tree(args)
            return [TextContent(type="text", text=tree)]
        case _:
            raise ValueError(f"Unknown tool: {name}")

以下是完整的 server.py 文件，其中详细说明了与 GitHub API 接口的 MCP 服务器的最小组件。

import base64
import json
import logging
import os
from enum import Enum
import requests
from dotenv import load_dotenv
from mcp.server import Server
from mcp.types import TextContent, Tool
from schemas import GetFileContentsArgs, GetTreeArgs, GitHubContent, GitHubTreeResponse
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("github-server")
logger.info("GitHub server started.")
load_dotenv()
app = Server("github_custom")

class GitHubTools(str, Enum):
    """Tool box definitions."""
    GET_FILE_CONTENTS = "get_file_contents"
    LIST_REPO_TREE = "list_repo_tree"

def get_file_contents(args: GetFileContentsArgs) -> GitHubContent:
    """Get the contents of a file from a GitHub repository."""
    url = f"https://api.github.com/repos/{args.owner}/{args.repo}/contents/{args.path}"
    if args.branch:
        url += f"?ref={args.branch}"
    headers = {
        "Authorization": f"token {os.environ.get('GITHUB_PERSONAL_ACCESS_TOKEN')}",
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "github-mcp-server",
    }
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    content = GitHubContent.model_validate(resp.json())
    if content.content:
        content.content = base64.b64decode(content.content).decode("utf-8")
    return content

def list_repo_tree(args: GetTreeArgs) -> str:
    """List the tree structure of a GitHub repository."""
    url = (
        f"https://api.github.com/repos/{args.owner}/{args.repo}/git/trees/"
        f"{args.branch}?recursive={args.recursive}"
    )
    logger.info(f"Listing tree for: {url}")
    headers = {
        "Authorization": f"token {os.environ.get('GITHUB_PERSONAL_ACCESS_TOKEN')}",
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "github-mcp-server",
    }
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    content = GitHubTreeResponse.model_validate(resp.json())
    return json.dumps(content.model_dump(mode="json"))

@app.list_tools()
async def list_tools() -> list[Tool]:
    """List available weather tools."""
    return [
        Tool(
            name="get_file_contents",
            description="Retrieve the file contents of a file from a github repository",
            inputSchema=GetFileContentsArgs.model_json_schema(),
        ),
        Tool(
            name="list_repo_tree",
            description="List the tree structure of a GitHub repository",
            inputSchema=GetTreeArgs.model_json_schema(),
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle the call_tool for the vector store service."""
    match name:
        case GitHubTools.GET_FILE_CONTENTS:
            args = GetFileContentsArgs(**arguments)
            file = get_file_contents(args)
            return [TextContent(type="text", text=file.content)]
        case GitHubTools.LIST_REPO_TREE:
            args = GetTreeArgs(**arguments)
            tree = list_repo_tree(args)
            return [TextContent(type="text", text=tree)]
        case _:
            raise ValueError(f"Unknown tool: {name}")

async def main() -> None:
    """Main entry point for server."""
    # Import here to avoid issues with event loops
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

测试你的服务器

MCP python-sdk 提供了一个客户端接口，用于连接到 MCP 服务器。在构建你自己的 MCP 服务器时，它有助于测试和调试工作负载。

from langchain_mcp_connect.data_models import StdioServerParameters
from mcp import ClientSession
from mcp.client.stdio import stdio_client
# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="python",
    args=["src/mcp_services/github"],
    env={"GITHUB_PERSONAL_ACCESS_TOKEN": "ENV_GITHUB_PERSONAL_ACCESS_TOKEN"},
)
async def run() -> None:
    """Run the client."""
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            # List available tools
            tools = await session.list_tools()
            print(tools)
            # Call a tool
            call_result = await session.call_tool(
                "get_file_contents",
                arguments={
                    "owner": "modelcontextprotocol",
                    "repo": "python-sdk",
                    "path": "LICENSE",
                },
            )
            print(call_result)
if __name__ == "__main__":
    import asyncio
    asyncio.run(run())

运行脚本的命令为：

python src/debug_server.py

如果你成功了，你将在控制台中看到此许可证的原始内容被打印出来。

meta=None content=[TextContent(type='text', text='MIT License\n\nCopyright (c) 2024 Anthropic, PBC\n\nPermission is hereby granted, free of charge, to any person obtaining a copy\nof this software and associated documentation files (the "Software"), to deal\nin the Software without restriction, including without limitation the rights\nto use, copy, modify, merge, publish, distribute, sublicense, and/or sell\ncopies of the Software, and to permit persons to whom the Software is\nfurnished to do so, subject to the following conditions:\n\nThe above copyright notice and this permission notice shall be included in all\ncopies or substantial portions of the Software.\n\nTHE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\nIMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\nFITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\nAUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\nLIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\nOUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\nSOFTWARE.\n')] isError=False'text', text='MIT License\n\nCopyright (c) 2024 Anthropic, PBC\n\nPermission is hereby granted, free of charge, to any person obtaining a copy\nof this software and associated documentation files (the "Software"), to deal\nin the Software without restriction, including without limitation the rights\nto use, copy, modify, merge, publish, distribute, sublicense, and/or sell\ncopies of the Software, and to permit persons to whom the Software is\nfurnished to do so, subject to the following conditions:\n\nThe above copyright notice and this permission notice shall be included in all\ncopies or substantial portions of the Software.\n\nTHE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\nIMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\nFITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\nAUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\nLIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\nOUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\nSOFTWARE.\n')] isError=False

环境变量

管理密钥是任何项目的关键方面。langchain_mcp_connect 库允许用户将当前环境中的密钥注入到服务器上下文中。

为此，在服务器参数中，将你的环境变量名称加上 ENV_ 前缀，以将环境变量注入到当前上下文中。

在上面的示例中，ENV_GITHUB_PERSONAL_ACCESS_TOKEN 值将主机环境中的 GITHUB_PERSONAL_ACCESS_TOKEN 值注入到 MCP 服务器环境中。

因此，要使上面的代码正常工作，你需要在当前环境中定义 GITHUB_PERSONAL_ACCESS_TOKEN，方法如下：

export GITHUB_PERSONAL_ACCESS_TOKEN="<YOUR_TOKEN_HERE>"

将 MCP 服务器连接到 LangChain

我们将利用 LangChain 工具调用库来与我们的 MCP 服务器集成。

首先，你需要在根目录中定义一个 claude_mcp_config.json 文件。此文件为 MCP 配置启动任何服务器所需的命令、参数和环境变量。你会发现配置中定义了多个 MCP 服务器。除了我们的自定义 GitHub 服务器外，MCP 还提供了对各种社区维护的服务器的访问。通过采用 MCP 标准，我们的项目可以轻松集成到这个不断增长的工具生态系统中。

{
  "mcpServers": {
    "github_custom": {
      "command": "python",
      "args": ["src/mcp_services/github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ENV_GITHUB_PERSONAL_ACCESS_TOKEN"
      }
    },
    "git": {
      "command": "uvx",
      "args": ["mcp-server-git", "--repository", "./"]
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "./"
      ]
    },
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ENV_GITHUB_PERSONAL_ACCESS_TOKEN"
      }
    }
  }
}

确保你已在当前环境中定义了 OPENAI_API_KEY。然后定义 langchain 的 OpenAI 集成。

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
  model="gpt-4o",
  model_kwargs={
      "max_tokens": 4096,
      "temperature": 0.0,
  },
)

从当前的 MCP 上下文中获取所有可用工具。

from langchain_mcp_connect import LangChainMcp
# Fetch tools
mcp = LangChainMcp()
tools = mcp.list_mcp_tools()

调用带有链接 MCP 工具的 LangChain 代理。

async def invoke_agent(invoke_agent(
    model: ChatOpenAI,
    query: str,
    langchain_tools: list[BaseTool],
) -> dict:
    """Invoke the agent with the given query."""
    agent_executor = create_react_agent(model, langchain_tools)
    # Create a system prompt and a human message
    human_message = HumanMessage(content=query)
    # Invoke the agent
    r = await agent_executor.ainvoke(input=dict(messages=[human_message]))
    return r
# Invoke the agent
response = asyncio.run(
    invoke_agent(model=llm, query=args.query, langchain_tools=tools)
)
print(response)

完整代码如下：

# /// script
# dependencies = [
#   "langchain>=0.3.9",
#   "langgraph>=0.2.53",
#   "langchain-openai>=0.2.10",
#   "langchain-community>=0.3.9",
#   "langchain-mcp-connect>=2.0.2",
# ]
# ///
import argparse
import asyncio
import logging
import os
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_core.tools import BaseTool
from langchain_mcp_connect import LangChainMcp
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
load_dotenv()
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("LangChainMcp")
if "GITHUB_PERSONAL_ACCESS_TOKEN" not in os.environ:
    raise ValueError(
        "Please set the GITHUB_PERSONAL_ACCESS_TOKEN environment variable."
    )
if "OPENAI_API_KEY" not in os.environ:
    raise ValueError("Please set the OPENAI_API_KEY environment variable.")

async def invoke_agent(
    model: ChatOpenAI,
    query: str,
    langchain_tools: list[BaseTool],
) -> dict:
    """Invoke the agent with the given query."""
    agent_executor = create_react_agent(model, langchain_tools)
    # Create a system prompt and a human message
    human_message = HumanMessage(content=query)
    # Invoke the agent
    r = await agent_executor.ainvoke(input=dict(messages=[human_message]))
    return r

if __name__ == "__main__":
    # Parse arguments
    parser = argparse.ArgumentParser(
        description="Langchain Model Context Protocol demo"
    )
    parser.add_argument(
        "-q",
        "--query",
        type=str,
        help="Query to be executed",
        default="What tools do you have access to?",
    )
    args = parser.parse_args()
    # Define the llm
    llm = ChatOpenAI(
        model="gpt-4o",
        model_kwargs={
            "max_tokens": 4096,
            "temperature": 0.0,
        },
    )
    # Fetch tools
    mcp = LangChainMcp()
    tools = mcp.list_mcp_tools()
    # Invoke the agent
    response = asyncio.run(
        invoke_agent(model=llm, query=args.query, langchain_tools=tools)
    )
    log.info(response)

此脚本可以在项目中通过以下方式执行：

uv run src/app/agent.py

结论

在模型上下文协议（MCP）引入之前，工具调用就已经存在。然而，在 MCP 出现之前，可用的工具生态系统缺乏标准化，这不利于在 AI 系统之间的可移植性。

双向大型语言模型（LLM）通信的标准化使用户能够创建、共享并推动一个由社区维护的丰富工具生态系统。MCP 为 AI 开发引入了一种模块化方法，其中后端工具可以快速集成，以创建定制化的 AI 功能。