MCP 与 Agent 框架集成：工具调用编排实战

你是一个能调用外部工具执行任务的智能体。可用工具如下：

{工具清单，例如：
- search_docs(查询语句): 检索技术文档
- query_sql(查询语句): 查询业务数据库
- send_email(收件人, 内容): 发送邮件}

请严格按以下格式思考与行动：
Thought: 你当前的思考
Action: 工具名（必须来自上面清单）
Action Input: {"参数名": "参数值"}
Observation: 工具返回结果
……（可多轮）
Thought: 我已得到答案
Final Answer: 最终回复

from mcp import ClientSession, StdioServerParameters
import asyncio, json

async def tools_to_openai_schema(session: ClientSession):
    """把 MCP 工具列表转换为 OpenAI 的 tools 参数"""
    result = await session.list_tools()
    return [
        {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description or "",
                "parameters": tool.inputSchema,   # JSON Schema 直接可用
            },
        }
        for tool in result.tools
    ]

from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.agents import create_react_agent
from langchain_openai import ChatOpenAI

async def build_agent():
    # 1. 用 MCP 客户端连接服务器（stdio 传输）
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "tsx", "/path/to/mcp-server/src/index.ts"],
    )

    # 2. 加载 MCP 工具为 LangChain Tool
    tools = await load_mcp_tools(server_params)

    # 3. 组装 ReAct Agent
    model = ChatOpenAI(model="gpt-4o", temperature=0)
    prompt = hub.pull("hwchase17/react")
    agent = create_react_agent(model, tools, prompt)

    # 4. 调用
    result = await agent.ainvoke({"input": "查询北京当前天气"})
    print(result["output"])

# 手动转换：理解底层映射
async def mcp_tool_to_langchain(name, mcp_tool):
    def invoke(args_str: str):
        args = json.loads(args_str)
        result = asyncio.run(session.call_tool(name, args))
        if result.isError:
            raise ToolException(result.content[0].text)
        return result.content[0].text
    return Tool(name=name, func=invoke, description=mcp_tool.description)

from llama_index.tools.mcp import BasicMCPClient, McpToolSpec
from llama_index.core.agent import FunctionCallingAgentWorker

def build_llama_index_agent():
    # 连接 MCP 服务器
    mcp_client = BasicMCPClient(
        "npx -y tsx /path/to/mcp-server/src/index.ts"
    )
    tool_spec = McpToolSpec.from_client(mcp_client)
    tools = tool_spec.to_tool_list()

    # 构建 Function Calling Agent
    agent = FunctionCallingAgentWorker.from_tools(
        tools,
        llm=OpenAI(model="gpt-4o"),
        verbose=True,
    ).as_agent()

    return agent

def should_confirm(tool_call, threshold=0.7) -> bool:
    """基于 logprob 判断工具选择是否可信"""
    conf = sum(arg.finish_reason == "tool_calls"
               for arg in [tool_call])  # 简化示意
    # 实际可读取 API 返回的 token logprob
    avg_prob = tool_call.get("avg_logprob", 0.0)
    return avg_prob < threshold

interface AgentStep {
  stepId: string;
  tool: string;
  args: Record<string, unknown>;
  result: string;         // 精简后的结果
  createdAt: number;
  tokensUsed: number;
}

class AgentMemory {
  private steps: AgentStep[] = [];

  append(step: AgentStep): void {
    this.steps.push(step);
    // 裁剪最旧步骤，防止上下文无限膨胀
    while (this.totalTokens() > MAX_STEP_TOKENS) {
      this.steps.shift();
    }
  }

  toContext(): string {
    // 只保留「结论性摘要」而非全部原始结果
    return this.steps
      .map((s) => `Step${s.stepId} [${s.tool}]: ${summarize(s.result)}`)
      .join("\n");
  }
}

# 第二步的参数引用第一步的结果
step1 = await agent.call_tool("query_orders", {"user_id": 42})
order_total = extract_amount(step1.result)

step2 = await agent.call_tool(
    "compute_discount", {"amount": order_total, "tier": "gold"}
)

step3 = await agent.call_tool(
    "send_email",
    {"to": "user@example.com", "body": f"您的折扣后金额为 {step2.result}"},
)

{
  "mcpServers": {
    "weather": {
      "command": "npx",
      "args": ["-y", "tsx", "/path/to/mcp-server/src/index.ts"]
    },
    "db": {
      "url": "https://mcp.example.com/mcp",
      "headers": { "Authorization": "Bearer <token>" }
    }
  }
}

from openai import OpenAI

client = OpenAI()

def run_with_mcp_tools(messages, tools):
    while True:
        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
        )
        choice = resp.choices[0]
        if not choice.message.tool_calls:
            return choice.message.content

        messages.append(choice.message)          # 带上 assistant 的 tool_calls
        for call in choice.message.tool_calls:
            # 经 MCP 客户端执行工具
            mcp_result = asyncio.run(
                session.call_tool(call.function.name,
                                  json.loads(call.function.arguments))
            )
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": mcp_result.content[0].text,
            })