核心摘要 ：Tools 是 Agent 从“聊天机器人”进化为“任务执行者”的关键。本文完整梳理 Tools 的请求/响应报文结构、参数定义与校验规范、工具代码执行机制、Function Calling 与 MCP 协议的本质区别，以及工具生态构建策略，助你构建稳定可靠的 Agent 工具链。

在 Agent 的四大核心组件中，Tools（工具） 是最具工程价值的部分。如果说 Planning 是 Agent 的“大脑”，Memory 是“记忆”，Action 是“肌肉”，那么 Tools 就是它的“手脚”。但很多开发者在实践中发现：请求报文拼不对、参数校验缺失、执行环境不安全、响应结果撑爆上下文 ，这些问题远比调 Prompt 更棘手。本文将从工程视角，系统拆解 Tools 的全链路实现，覆盖原文所有核心知识点并补充落地细节。

Agent Tools 知识全景图

在深入细节前，我们先建立一个全局认知。Agent Tools 体系可以分为四个层次：

层级核心内容关键问题
第一层：核心机制Function Calling 协议LLM 与代码如何通信？数据格式是什么？
第二层：工具定义Schema 设计技巧如何让 LLM 准确理解并正确调用工具？
第三层：工具执行代码实现与循环控制如何解析、执行、处理错误及多轮调用？
第四层：工具生态MCP 协议如何实现工具的标准化接入与跨平台复用？

一、核心机制：LLM 与代码的“通信语言”

Function Calling 本质上是 LLM 与外部代码之间的一套结构化通信协议。它定义了数据怎么传、格式是什么样的。

1.1 六步循环：完整交互流程图解

理解 Function Calling，首先要记住这个核心循环：

┌─────────────────────────────────────────┐
│ Step 1: 系统初始化                        │
│ 代码发送：系统提示 + 工具列表（Schema）     │
│ 告诉 LLM：“你有这些工具可用”               │
└──────────────────┬──────────────────────┘
                   ↓
┌─────────────────────────────────────────┐
│ Step 2: 用户输入                          │
│ 用户说："北京今天天气怎么样？"              │
└──────────────────┬──────────────────────┘
                   ↓
┌─────────────────────────────────────────┐
│ Step 3: LLM 决策                         │
│ LLM 内部推理："需要查天气"                 │
│ 输出：工具调用请求（JSON）                  │
│ ⚠️ 注意：不是直接回答，是"我要调工具"       │
└──────────────────┬──────────────────────┘
                   ↓
┌─────────────────────────────────────────┐
│ Step 4: 代码解析执行                      │
│ 代码解析 JSON → 找到对应函数               │
│ 执行真实工具 → 获取结果                    │
│ 如：调用天气 API → 返回 {"temp": 25}      │
└──────────────────┬──────────────────────┘
                   ↓
┌─────────────────────────────────────────┐
│ Step 5: 结果返回给 LLM                    │
│ 代码把工具结果包装成特定格式                │
│ 发送回 LLM，让它基于真实数据回答            │
└──────────────────┬──────────────────────┘
                   ↓
┌─────────────────────────────────────────┐
│ Step 6: LLM 最终回答                     │
│ LLM 基于工具结果生成自然语言回答            │
│ "北京今天晴天，25度，适合出门"              │
└─────────────────────────────────────────┘

💡 三个关键特征

  • 循环性：Step 3-5 可能循环多次（复杂任务需要多轮工具调用）
  • 分离性：LLM 只“说”要做什么，代码真正“做”
  • 状态性：每次调用都有唯一 ID，用于追踪请求与响应的对应关系

1.2 工具描述 Schema：告诉 LLM “你有什么”

这是 Function Calling 的起点。一个标准的工具定义如下：

{
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "获取指定城市的当前天气情况，包括温度、湿度和天气状况",
        "parameters": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "城市名称，例如：北京、Shanghai、London"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "温度单位，默认为celsius"
            }
          },
          "required": ["city"]
        }
      }
    }
  ],
  "tool_choice": "auto"
}

字段速查表：

字段路径类型必填含义注意事项
tools[].function.namestring✅函数名a-z0-9_，不以数字开头
tools[].function.descriptionstring✅功能描述LLM 靠这个理解用途，越清晰越好
tools[].function.parametersobject✅参数定义标准 JSON Schema 格式

parameters 结构详解

{
  "type": "object",
  "properties": {
    "参数名": {
      "type": "string/number/integer/boolean/array/object",
      "description": "参数说明",
      "enum": ["可选", "限制可选值"],
      "default": "可选默认值"
    }
  },
  "required": ["需要必填的参数名"]
}

关键规则

  • type: "object" 固定，表示参数整体是一个对象
  • properties 里定义每个具体参数
  • required 是字符串数组，列出必填参数名
  • 参数类型：string, number, integer, boolean, array, object

1.3 LLM 调用请求与响应格式

当 LLM 决定调用工具时，它会返回 tool_calls 而非普通文本：

{
  "id": "chatcmpl-abc123",
  "object": "chat.completion",
  "created": 1712345678,
  "model": "glm-5",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"city\": \"北京\", \"unit\": \"celsius\"}"
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

⚠️ 高频踩坑点arguments 字段是 JSON 字符串，不是对象！使用前必须 JSON.parse()。

关键字段

字段类型含义注意
message.contentnull/string通常为 nullnull 表示要调工具，有值表示直接回答
message.tool_callsarray工具调用列表可能同时调多个
tool_calls[].idstring唯一标识后续返回必须原样带回
tool_calls[].typestring固定 "function"—
tool_calls[].function.namestring要调用的函数名必须在 tools 列表里
tool_calls[].function.argumentsstring参数 JSON 字符串⚠️ 不是对象，要 JSON.parse

多工具并行示例

{
  "tool_calls": [
    {
      "id": "call_1",
      "function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}
    },
    {
      "id": "call_2",
      "function": {"name": "search_web", "arguments": "{\"query\": \"AI趋势\"}"}
    }
  ]
}

代码处理：遍历数组，并行或串行执行，分别返回结果。

1.4 工具结果返回与 tool_choice 控制

返回格式：使用 role: "tool" 消息，且 tool_call_id 必须与原请求一致：

{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "name": "get_weather",
  "content": "{\"temp\": 25, \"weather\": \"晴\", \"humidity\": 60}"
}

💡 错误处理原则：即使工具执行失败，也要返回错误信息（而非抛异常），让 LLM 自行决定重试、换工具还是告知用户。

tool_choice 控制策略：

取值含义适用场景
"auto"LLM 自主决定通用对话，不确定是否需要工具
"required" / "any"必须调用至少一个工具明确需要外部数据，禁止瞎编
"none"禁止调用工具纯文本生成、测试 LLM 本身能力
{"function": {"name": "xxx"}}强制指定工具工作流中明确知道下一步用什么

二、Schema 设计技巧：让 LLM 精准调用

工具定义得好不好，直接决定了 Agent 的稳定性。这一层属于“设计层”。

2.1 基础与复杂参数类型

除了 string/number/integer/boolean 四种基础类型，还需掌握：

  • 数组参数（批量处理）：使用 items 定义元素类型，配合 minItems/maxItems 约束数量
  • 嵌套对象（结构化数据）：支持多层嵌套，但 建议不超过 3 层，否则 LLM 容易混乱
  • 枚举约束：用 enum 限制可选值，用 minimum/maximum 限制数值范围

2.1.2 示例展示

  • 基础示例
{
  "name": "send_message",
  "parameters": {
    "type": "object",
    "properties": {
      "to": {"type": "string", "description": "收件人"},
      "content": {"type": "string", "description": "消息内容"},
      "urgent": {"type": "boolean", "description": "是否紧急", "default": false}
    },
    "required": ["to", "content"]
  }
}
  • 数组参数（批量处理）
{
  "name": "send_emails",
  "description": "批量发送邮件",
  "parameters": {
    "type": "object",
    "properties": {
      "recipients": {
        "type": "array",
        "description": "收件人邮箱列表，至少1个",
        "items": {
          "type": "string",
          "description": "邮箱地址，如 user@example.com",
          "format": "email"
        },
        "minItems": 1,
        "maxItems": 50
      },
      "subject": {"type": "string", "description": "邮件主题"},
      "template_id": {"type": "string", "description": "邮件模板ID"}
    },
    "required": ["recipients", "subject"]
  }
}
  • 嵌套对象（结构化数据）
{
  "name": "create_user",
  "parameters": {
    "type": "object",
    "properties": {
      "name": {"type": "string"},
      "contact": {
        "type": "object",
        "description": "联系方式",
        "properties": {
          "phone": {"type": "string"},
          "email": {"type": "string"}
        },
        "required": ["phone"]
      }
    }
  }
}
  • 枚举约束（限制选项）

字符串枚举

{
  "unit": {
    "type": "string",
    "enum": ["celsius", "fahrenheit", "kelvin"],
    "description": "温度单位"
  }
}

整数枚举(配合 minimum / maximum)

{
  "priority": {
    "type": "integer",
    "minimum": 1,
    "maximum": 5,
    "description": "优先级，1=最低，5=最高"
  }
}

布尔 + 默认值

{
  "include_details": {
    "type": "boolean",
    "description": "是否包含详细信息",
    "default": false
  }
}

2.2 高级约束：anyOf vs oneOf

当参数可以是多种类型时（例如 query 可以是字符串或对象），使用组合约束：

约束关键字用途
字符串长度minLength, maxLength密码、验证码
数值范围minimum, maximum, exclusiveMinimum年龄、价格
正则匹配pattern手机号、邮箱格式
数组唯一uniqueItems: true去重列表
多类型anyOf, oneOf参数可传字符串或对象

字符串长度(minLength / maxLength)

{
  "name": "create_user",
  "parameters": {
    "type": "object",
    "properties": {
      "username": {
        "type": "string",
        "minLength": 3,
        "maxLength": 20,
        "description": "用户名，3-20个字符"
      },
      "password": {
        "type": "string",
        "minLength": 8,
        "maxLength": 20,
        "description": "密码，8-20个字符"
      }
    },
    "required": ["username", "password"]
  }
}

数值范围

{
  "name": "create_product",
  "parameters": {
    "type": "object",
    "properties": {
      "discount_percent": {
        "type": "integer",
        "minimum": 0,
        "maximum": 100,
        "description": "折扣百分比，0-100"
      },
      "temperature": {
        "type": "number",
        "exclusiveMinimum": -273.15,
        "description": "温度，必须大于绝对零度（-273.15）"
      }
    }
  }
}

正则示例

{
  "phone": {
    "type": "string",
    "pattern": "^1[3-9]\\d{9}$",
    "description": "中国大陆手机号"
  }
}

数组唯一

{
  "name": "tag_user",
  "parameters": {
    "type": "object",
    "properties": {
      "user_id": {"type": "string"},
      "tags": {
        "type": "array",
        "items": {
          "type": "string",
          "description": "标签名，如VIP、活跃用户"
        },
        "minItems": 1,
        "maxItems": 10,
        "uniqueItems": true,
        "description": "标签列表，1-10个，不能重复"
      }
    }
  }
}

多类型

用于参数可以是多种类型的，例如日常的 query 参数，可以传入字符串或者对象。

多类型又分两种，一种是anyOf满足某一个类型就可，一种是oneOf 只能满足一个类型。

例如两个条件字符串长度 >3 或者 <6，假如传入“abcd” ，他两个条件都满足，对于 anyOf 类型的它就能通过，oneOf 的就不行，必须是“ab” 只匹配了<6，oneOf 才允许。

anyOf

{
  "name": "make_payment",
  "parameters": {
    "type": "object",
    "properties": {
      "amount": {
        "anyOf": [
          {
            "type": "number",
            "description": "金额数字，单位元，如 100.5"
          },
          {
            "type": "object",
            "description": "金额对象，含币种",
            "properties": {
              "value": {"type": "number"},
              "currency": {"type": "string", "enum": ["CNY", "USD"]}
            },
            "required": ["value", "currency"]
          }
        ]
      }
    }
  }
}
输入结果说明
100✅ 通过匹配数字 schema
{"value": 100, "currency": "CNY"}✅ 通过匹配对象 schema
{"value": 100}❌ 失败不匹配任何 schema（缺 currency）

oneOf

{
  "oneOf": [
    {"type": "string", "pattern": "^[0-9]+$"},      // 纯数字字符串
    {"type": "string", "pattern": "^[a-zA-Z]+$"}     // 纯字母字符串
  ]
}
输入anyOfoneOf原因
“ab”✅✅只匹配 maxLength≤5
“abcdef”✅✅只匹配 minLength≥3
“hello”✅❌同时匹配两个！长度=5，≥3且≤5

⚠️ 区别提醒：如果传入的值同时满足多个子 schema，anyOf 通过，oneOf 会失败。

2.3 描述优化的 5 条黄金法则

原则❌ 差的写法✅ 好的写法
动词开头weather_dataget_weather
具体明确处理数据查询订单状态，返回物流信息和预计到达时间
给出示例城市名称城市名称，如"北京"、"New York"
说清限制时间参数格式YYYY-MM-DD，默认今天，最早2020-01-01
默认值明确可选默认为true，表示包含已删除订单

三、代码实战：从零实现 Function Calling

理论讲完，我们用一个完整的 Python 示例跑通全流程。

3.1 编写工具函数

import requests  
  
def get_weather(city: str) -> str:  
    """  
    通过调用 wttr.in API 查询真实的天气信息。  
    """    url = f"https://wttr.in/{city}?format=j1"  
  
    try:  
        response = requests.get(url)  
        response.raise_for_status()  
        data = response.json()  
  
        current_condition = data['current_condition'][0]  
        weather_desc = current_condition['weatherDesc'][0]['value']  
        temp_c = current_condition['temp_C']  
  
        return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"  
  
    except requests.exceptions.RequestException as e:  
        return f"错误:查询天气时遇到网络问题 - {e}"  
    except (KeyError, IndexError) as e:  
        return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"

3.2 封装通用 LLM 客户端

from openai import OpenAI  
  
class OpenAICompatibleClient:  
    """  
    一个用于调用任何兼容OpenAI接口的LLM服务的客户端。  
    """  
    def __init__(self, model: str, api_key: str, base_url: str):  
        self.model = model  
        self.client = OpenAI(api_key=api_key, base_url=base_url)  
  
    def generate(  
        self,  
        messages: list,  
        tools: list | None = None,  
        tool_choice: str | None = "auto",  
        system_prompt: str | None = None  
    ):  
        """  
        调用LLM API，支持Function Calling。  
  
        Args:            
	        messages: 消息历史列表  
            tools: 工具定义列表 (Function Calling格式)  
            tool_choice: 工具选择策略 ("auto", "none", 或指定函数名)  
            system_prompt: 系统提示词  
  
        Returns:            
	        LLM的响应，包含function_call信息（如果有）  
        """        
        print("正在调用大语言模型...")  
  
		"""如果有系统提示词就进行拼接"""
        if system_prompt:  
            full_messages = [{"role": "system", "content": system_prompt}] + messages  
        else:  
            full_messages = messages  
  
		"""连接 LLM """
        try:  
            response = self.client.chat.completions.create(  
                model=self.model,  
                messages=full_messages,  
                tools=tools,  
                tool_choice=tool_choice,  
                stream=False  
            )  
  
            message = response.choices[0].message  
            print("大语言模型响应成功。")  
  
            return message  
  
        except Exception as e:  
            print(f"调用LLM API时发生错误: {e}")  
            return None

3.3 定义工具

available_functions = {  
    "get_weather": get_weather
}  
  
TOOL_DEFINITIONS = [  
    {  
        "type": "function",  
        "function": {  
            "name": "get_weather",  
            "description": "查询指定城市的实时天气信息",  
            "parameters": {  
                "type": "object",  
                "properties": {  
                    "city": {  
                        "type": "string",  
                        "description": "城市名称，如：北京、上海、深圳"  
                    }  
                },  
                "required": ["city"]  
            }  
        }  
    }  
]

3.4 装载与调用

import json  
  
from tools_call.core import OpenAICompatibleClient, API_KEY, BASE_URL, MODEL_ID  
from tools_call.tools import available_functions, TOOL_DEFINITIONS  
  
  
def run_function_calling_agent(user_input: str, max_turns: int = 10):  
    """  
    Function Calling 演示 demo      
    Args:  
        user_input: 用户输入  
        max_turns: 最大循环次数  
    """   
    
	"""初始化客户端，传入 model id，api key，大模型url """ 
	llm = OpenAICompatibleClient(  
        model=MODEL_ID,  
        api_key=API_KEY,  
        base_url=BASE_URL  
    )  
      
    messages = [{"role": "user", "content": user_input}]  
    custom_tool_names = list(available_functions.keys())  
      
    print(f"用户输入: {user_input}\n" + "=" * 40)  
    print(f"已加载工具: {custom_tool_names}\n")  
      
    for turn in range(max_turns):  
        print(f"--- 循环 {turn + 1} ---\n")  
          
        """连接 llm，并告诉 llm 有哪些工具可调用 """  
        response = llm.generate(  
            messages=messages,  
            tools=TOOL_DEFINITIONS,  
            tool_choice="auto"  
        )  
          
        if response is None:  
            print("LLM调用失败，退出循环")  
            break  
        if response.content:  
            print(f"模型文本回复:\n{response.content}\n")  
          
        if response.tool_calls:  
            print(f"模型调用了 {len(response.tool_calls)} 个工具:")  
              
            for tool_call in response.tool_calls:  
                tool_name = tool_call.function.name  
                tool_args = json.loads(tool_call.function.arguments)  
                  
                print(f"\n工具名: {tool_name}")  
                print(f"参数: {tool_args}")  
                  
                if tool_name in custom_tool_names:  
                    try:  
                        result = available_functionstool_name  
                    except Exception as e:  
                        result = json.dumps({"error": type(e).__name__, "message": str(e)})  
                else:  
                    result = json.dumps({"error": "UnknownTool", "message": f"未定义的工具 '{tool_name}'"})  
                  
                print(f"执行结果: {result}")  
                
				""" 将工具的执行结果组装，返回给 llm, 让它进行总结输出 """	  
                messages.append({  
                    "role": "assistant",  
                    "content": response.content,  
                    "tool_calls": [  
                        {  
                            "id": tool_call.id,  
                            "type": "function",  
                            "function": {  
                                "name": tool_name,  
                                "arguments": json.dumps(tool_args)  
                            }  
                        }  
                    ]  
                })  
                  
                messages.append({  
                    "role": "tool",  
                    "tool_call_id": tool_call.id,  
                    "content": str(result)  
                })  
              
            print("\n" + "=" * 40)  
        else:  
            print("模型没有调用工具，结束对话。")  
            break  
        return None  
  
  
if __name__ == "__main__":  
    user_input = "你好，请帮我查询一下今天深圳的天气。"  
    run_function_calling_agent(user_input)

💡 终止条件：LLM 返回 content 有值且 tool_calls 为空，或达到最大轮数（防止无限循环）。

四、进阶：MCP 协议实现工具生态标准化

当工具越来越多、跨项目复用时，Function Calling 的硬编码方式就不够用了。MCP（Model Context Protocol） 提供了标准化的工具发现与调用协议。

4.1 MCP 协议实现

4.1.1 编写一个简单获取天气的 mcp server

import asyncio
import requests
from mcp.server import Server
from mcp.types import Tool, TextContent

def get_weather(city: str) -> str:
    """获取天气"""
    url = f"https://wttr.in/{city}?format=j1"
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        data = response.json()
        current = data['current_condition'][0]
        weather = current['weatherDesc'][0]['value']
        temp = current['temp_C']
        return f"{city}当前天气:{weather}，气温{temp}摄氏度"
    except Exception as e:
        return f"获取天气失败:{e}"

mcp = Server("mcpServer")

@mcp.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="获取城市天气",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称，如：北京、上海、深圳"}
                },
                "required": ["city"]
            },
        )
    ]

@mcp.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_weather":
        weather = get_weather(arguments["city"])
        return [TextContent(type="text", text=weather)]
    raise ValueError(f"不支持的工具:{name}")

async def main():
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await mcp.run(
            read_stream,
            write_stream,
            mcp.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

4.1.2 编写连接 mcp server 的客户端

"""
MCP Client 模块

封装 MCP 协议客户端，支持连接多个 MCP Server"""

import os
import sys
from typing import Any
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters

class MCPClient:
    """单个 MCP Server 连接管理"""
    def __init__(self, server_script: str):
        # Server 脚本路径
        self.server_script = server_script
        # MCP 会话实例，通过它发送请求给 Server
        self._session: ClientSession | None = None
        # stdio 传输层的上下文管理器，用于正确清理资源
        self._stdio_ctx = None
        # 缓存已加载的工具列表
        self._tools: list[dict] = []

    async def connect(self):
		# 获取当前 Python 解释器路径，确保 Server 使用相同的环境
        python_exec = sys.executable

        # 配置 Server 启动参数
        server_params = StdioServerParameters(
            command=python_exec,  # Python 解释器
            args=[self.server_script],  # Server 脚本路径
            env=None
        )

        # stdio_client 是 async context manager，启动子进程并建立通信管道
        # __aenter__ 返回 (read_stream, write_stream)
        self._stdio_ctx = stdio_client(server_params)
        read, write = await self._stdio_ctx.__aenter__()

        # 创建 MCP ClientSession，它是与 Server 通信的核心
        self._session = ClientSession(read, write)
        # 进入 session 上下文
        await self._session.__aenter__()
        # 发送初始化请求，建立协议连接
        await self._session.initialize()
        # 从 Server 获取可用工具列表
        await self._load_tools()

    async def _load_tools(self):
        """从 Server 获取工具列表，转换为 Function Calling 格式"""
        if not self._session:
            raise RuntimeError("MCP Client not connected")

        # 调用 Server 的 list_tools 方法
        tools_response = await self._session.list_tools()
        self._tools = []

        # 遍历每个工具，转换为 LLM 理解的格式
        for tool in tools_response.tools:
            self._tools.append({
                "type": "function",
                "function": {
                    "name": tool.name, # 工具名
                    "description": tool.description or "", # 描述（供 LLM 判断）
                    "parameters": tool.inputSchema  # JSON Schema 参数定义
                }
            })

    async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> str:
        """
        调用 Server 上的工具
                Args:
            tool_name: 工具名称
            arguments: 工具参数
                Returns:
            工具执行结果的文本形式
        """
        if not self._session:
            raise RuntimeError("MCP Client not connected")

        # 通过 MCP 协议发送 call_tool 请求到 Server
        result = await self._session.call_tool(tool_name, arguments)

        # 提取返回的文本内容
        if result.content and len(result.content) > 0:
            return result.content[0].text
        return ""

    def get_tools(self) -> list[dict]:
        """获取已加载的工具定义列表"""
        return self._tools

    async def close(self):
        """关闭连接，清理资源"""
        # 关闭 ClientSession
        if self._session:
            await self._session.__aexit__(None, None, None)
            self._session = None
        # 关闭 stdio 传输层（会终止子进程）
        if self._stdio_ctx:
            await self._stdio_ctx.__aexit__(None, None, None)
            self._stdio_ctx = None

class MCPClientManager:
    """多 MCP Server 管理"""
    def __init__(self):
        # 存储多个 Server 的客户端实例，key 为 Server 名称
        self._clients: dict[str, MCPClient] = {}

    async def add_server(self, name: str, server_script: str):
        """
        添加并连接一个 MCP Server
        Args:
            name: Server 名称标识
            server_script: Server 脚本路径
        """
        client = MCPClient(server_script)
        await client.connect()
        self._clients[name] = client

    def get_tools(self) -> list[dict]:
        """获取所有 Server 上所有工具的合集"""
        all_tools = []
        for client in self._clients.values():
            all_tools.extend(client.get_tools())
        return all_tools

    async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> str:
        """
        根据工具名找到对应的 Server 并调用
        遍历所有已连接的 Server，查找提供该工具的 Server 并调用
        """
        for client in self._clients.values():
            tool_names = [t["function"]["name"] for t in client.get_tools()]
            if tool_name in tool_names:
                return await client.call_tool(tool_name, arguments)
        raise ValueError(f"Tool '{tool_name}' not found in any MCP server")

    async def close_all(self):
        """关闭所有 Server 连接"""
        for client in self._clients.values():
            await client.close()
        self._clients.clear()

4.1.3 装载并调用

import json
import asyncio
import os

from tools_call.approaches.mcp.client import MCPClientManager
from tools_call.core.config import MODEL_ID, API_KEY, BASE_URL
from tools_call.core.openai_client import OpenAICompatibleClient

async def run_mcp_agent(user_input: str, max_turns: int = 10):
    """
    运行 MCP 模式的 Agent
    Args:
        user_input: 用户输入
        max_turns: 最大循环次数
    """

    llm = OpenAICompatibleClient(
        model=MODEL_ID,
        api_key=API_KEY,
        base_url=BASE_URL
    )

    mcp_manager = MCPClientManager()

    server_config = {
        "demo": "/Volumes/xwbData/data/code/myself/python/Agent_demo/server.py"
    }

    for name, script in server_config.items():
        print(f"连接 MCP Server:{name}")
        await mcp_manager.add_server(name, script)

    mcp_tools = mcp_manager.get_tools()
    all_tools = mcp_tools

    messages = [{"role": "user", "content": user_input}]

    print(f"用户输入:{user_input}\n" + "=" * 40)
    print(f"MCP 工具:{[t['function']['name'] for t in mcp_tools]}")
    print(f"总工具数:{len(all_tools)}\n")

    try:
        for turn in range(max_turns):
            print(f"--- 循环{turn + 1} ---\n")

            response = llm.generate(
                messages=messages,
                tools=all_tools,
                tool_choice="auto"
            )

            if response is None:
                print("LLM调用失败，退出循环")
                break
            if response.content:
                print(f"模型文本回复:\n{response.content}\n")

            if response.tool_calls:
                print(f"模型调用了{len(response.tool_calls)} 个工具:")

                for tool_call in response.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = json.loads(tool_call.function.arguments)

                    print(f"\n工具名:{tool_name}")
                    print(f"参数:{tool_args}")

                    try:
                        result = await mcp_manager.call_tool(tool_name, tool_args)
                        tool_type = "MCP"
                    except Exception as e:
                        result = json.dumps({"error": type(e).__name__, "message": str(e)})
                        tool_type = "MCP"
                    print(f"执行结果:{result}")
                    print(f"工具类型:{tool_type}")

                    messages.append({
                        "role": "assistant",
                        "content": response.content,
                        "tool_calls": [
                            {
                                "id": tool_call.id,
                                "type": "function",
                                "function": {
                                    "name": tool_name,
                                    "arguments": json.dumps(tool_args)
                                }
                            }
                        ]
                    })

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })

                print("\n" + "=" * 40)
            else:
                print("模型没有调用工具，结束对话。")
                break
    finally:
        await mcp_manager.close_all()
        print("\nMCP Server 连接已关闭。")

if __name__ == "__main__":
    user_input = "你好，请帮我查询一下今天深圳的天气"
    asyncio.run(run_mcp_agent(user_input))
# 人工智能 # 学习方法 # AIGC
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

别让你的 Agent 一口吃成胖子：AI Agent 任务拆分的工程实践指南

avatar AI Agent技术社区
cover

AI Agent律师行业智能体的实践步骤

avatar AI Agent技术社区

MCP + 企业IM架构：AI Agent调用OA、ERP、MES时，身份和权限如何传递？

<think>我们只需要根据内容生成摘要，不超过150字。先理解文章核心：MCP+企业IM架构，AI Agent调用OA/ERP/MES时身份和权限传递。要点：IM从消息入口变任务入口；MCP解决连接，但不解决权限；需要五层身份映射；权限分层（可见性、调用、数据范围、确认）；建议组合MCP+API网关。摘要需简洁。 参考生成： "本文探讨MCP与企业IM集成中AI Agent调用业务系统时的身份权

avatar AI Agent技术社区