AutoGen MCP 简单使用

王学政2

739人浏览 · 2025-07-03 16:14:56

王学政2 · 2025-07-03 16:14:56 发布

安装MCP SDK

pip install mcp -i https://mirrors.aliyun.com/pypi/simple/

自定义MCPServer服务

calculator_server.py

import logging

from mcp.server import FastMCP
from mcp.types import TextContent

# 日志相关配置
logging.basicConfig(
    level=logging.INFO,
    encoding="utf-8",
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("calculator_mcp_server")

# 初始化FastMCP服务器，并指定服务的名称为：calculator
mcp_server = FastMCP(
    name="calculator_mcp_server", # 服务名
    instructions="计算器", # 服务描述
)

# 定义加法工具函数
@mcp_server.tool()
async def add(x: float, y: float) -> list[TextContent]:
    """
    加法运算
    :param x: 第一个数字
    :param y: 第二个数字
    :return:
    """
    res = x + y
    logger.info(f"加法运算 :{x} + {y} = {res}")
    return [TextContent(type="text", text=str(res))]

# 定义减法工具函数
@mcp_server.tool()
async def subtract(x: float, y: float) -> list[TextContent]:
    """
    减法运算
    :param x: 第一个数字
    :param y: 第二个数字
    :return:
    """
    res = x - y
    logger.info(f"减法运算:{x} - {y} = {res}")
    return [TextContent(type="text", text=str(res))]

# 定义乘法运算工具函数
@mcp_server.tool()
async def multiply(x: float, y: float) -> list[TextContent]:
    """
    乘法运算
    :param x: 第一个数字
    :param y: 第二个数字
    :return:
    """
    res = x * y
    logger.info(f"乘法运算 :{x} * {y} = {res}")
    return [TextContent(type="text", text=str(res))]

# 定义除法运算工具函数
@mcp_server.tool()
async def divide(x: float, y: float) -> list[TextContent]:
    """
    除法运算
    :param x: 第一个数字
    :param y: 第二个数字
    :return:
    """
    res = x / y
    logger.info(f"除法运算:{x} / {y} = {res}")
    return [TextContent(type="text", text=str(res))]


if __name__ == "__main__":
    # 以标准 I/O 方式运行 MCP 服务器
    mcp_server.run(transport='stdio')

客户端调用MCPServer

STDIO 标准输入输出方式

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.tools.mcp import StdioServerParams, mcp_server_tools

from my_llms_client.llms_client import LLMsClient

# 模型客户端
model_client = LLMsClient.qwen_client()

# STDIO MCPServer服务的配置
calculator_mcp_server = StdioServerParams(
    command="python",
    args=["calculator_server.py"]
)

# 运行智能体
async def run_agent(task: str):

    # 获取MCPServer提供的工具
    tools = await mcp_server_tools(calculator_mcp_server)

    #创建智能体
    agent = AssistantAgent(
        name="my_agent",
        model_client=model_client,
        tools=tools,
        reflect_on_tool_use=True,
        system_message="你是一个算术大师，能够使用工具进行算术运算",
    )

    """使用 Console 实时打印所有消息。"""
    await Console(
        agent.run_stream(task=task),
        output_stats=True,  # 启用统计信息打印。
    )

if __name__ == '__main__':

    task = "1 + 1 等于？"
    asyncio.run(run_agent(task=task))

运行结果：

2025-07-03 15:23:26,428 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest
---------- TextMessage (user) ----------
1 + 1 等于？
---------- ToolCallRequestEvent (my_agent) ----------
[FunctionCall(id='call_03f2fd66be694621b7ee90', arguments='{"x": 1, "y": 1}', name='add')]
[Prompt tokens: 550, Completion tokens: 22]
2025-07-03 15:23:29,060 - mcp.server.lowlevel.server - INFO - Processing request of type CallToolRequest
2025-07-03 15:23:29,063 - calculator_mcp_server - INFO - �ӷ����� :1.0 + 1.0 = 2.0
2025-07-03 15:23:29,073 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest
---------- ToolCallExecutionEvent (my_agent) ----------
[FunctionExecutionResult(content='[{"type": "text", "text": "2.0", "annotations": null, "meta": null}]', name='add', call_id='call_03f2fd66be694621b7ee90', is_error=False)]
---------- TextMessage (my_agent) ----------
1 + 1 等于 2。给出的答案 "2.0" 虽然是正确的，但通常在处理整数加法时，我们会简单地说结果是 2。这里的 ".0" 表示这是一个浮点数形式的2，但在这种情况下，更常见的表达方式是整数2。
[Prompt tokens: 60, Completion tokens: 70]
---------- Summary ----------
Number of messages: 4
Finish reason: None
Total prompt tokens: 610
Total completion tokens: 92
Duration: 5.66 seconds

Process finished with exit code 0

使用第三方MCP服务

示例：高德地图

高德地图开放平台: 高德开放平台 | 高德地图API

注册申请APIKEY

高德MCPServer: 概述-MCP Server | 高德地图API

客户端调用示例：

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.tools.mcp import SseServerParams, mcp_server_tools

from my_llms_client.llms_client import LLMsClient

# 模型客户端
model_client = LLMsClient.qwen_client()

# 高德地图SSE
gaode_server_params = SseServerParams(
    url="https://mcp.amap.com/sse?key=xxxxx"
)

# 运行智能体
async def run_agent(task: str):

    # 获取MCPServer提供的工具
    tools = await mcp_server_tools(gaode_server_params)

    #创建智能体
    agent = AssistantAgent(
        name="my_agent",
        model_client=model_client,
        tools=tools,
        reflect_on_tool_use=True,
        system_message="""
            你是一个LBS地图专家，能够使用工具完成一些关于地图、地理的相关功能。
        """,
    )

    """使用 Console 实时打印所有消息。"""
    await Console(
        agent.run_stream(task=task),
        output_stats=True,  # 启用统计信息打印。
    )

if __name__ == '__main__':

    task = ("公共交通出行规划："
            "从楼山后社区出发，坐标(120.391909,36.21425) 到 桃园雅居，坐标(120.435682,36.19846)")
    asyncio.run(run_agent(task=task))

运行结果：

---------- TextMessage (user) ----------
公共交通出行规划：从楼山后社区出发，坐标(120.391909,36.21425) 到 桃园雅居，坐标(120.435682,36.19846)
---------- ToolCallRequestEvent (my_agent) ----------
[FunctionCall(id='call_b1528ab68e304f9fb59996', arguments='{"origin": "120.391909,36.21425", "destination": "120.435682,36.19846", "city": "青岛市", "cityd": "青岛市"}', name='maps_direction_transit_integrated')]
[Prompt tokens: 2515, Completion tokens: 76]
---------- ToolCallExecutionEvent (my_agent) ----------
[FunctionExecutionResult(content='[...]}", "annotations": null, "meta": null}]', name='maps_direction_transit_integrated', call_id='call_b1528ab68e304f9fb59996', is_error=False)]
---------- TextMessage (my_agent) ----------
根据提供的信息，从楼山后社区（坐标：120.391909, 36.21425）到桃园雅居（坐标：120.435682, 36.19846）的公共交通出行规划有以下几种方案：

### 方案一
- **总时长**：约4250秒（约70分钟）
- **步行距离**：约1282米
- **具体步骤**：
  1. 步行至楼山路四流北路公交站。
  2. 乘坐22路公交车，在楼山路重庆中路站上车，于海头文登路站下车。
  3. 换乘13路公交车，在海头文登路站上车，于桃园站下车。
  4. 步行至目的地。

### 方案二
- **总时长**：约5402秒（约90分钟）
- **步行距离**：约799米
- **具体步骤**：
  1. 步行至胸科医院公交站。
  2. 乘坐105路公交车，在胸科医院站上车，于虎山路东站下车。
  3. 换乘123路公交车，在虎山路东站上车，于桃园雅居站下车。
  4. 步行至目的地。

### 方案三
- **总时长**：约5341秒（约89分钟）
- **步行距离**：约923米
- **具体步骤**：
  1. 步行至胸科医院公交站。
  2. 乘坐606路公交车，在胸科医院站上车，于夏庄路金水路站下车。
  3. 换乘123路公交车，在夏庄路金水路站上车，于桃园雅居站下车。
  4. 步行至目的地。

### 方案四
- **总时长**：约7333秒（约122分钟）
- **步行距离**：约534米
- **具体步骤**：
  1. 步行至威山后公交站。
  2. 乘坐38路或115路公交车，在威山后站上车，于升平路小区站下车。
  3. 换乘387路公交车，在升平路小区站上车，于下王埠东站下车。
  4. 再次换乘123路公交车，在下王埠东站上车，于桃园雅居站下车。
  5. 步行至目的地。

以上方案均考虑了步行和换乘的时间。建议根据实际情况选择最合适的路线。如果时间充裕且不介意多走几步，可以选择步行较少的方案；如果希望减少步行距离，则可以考虑步行较多但总耗时较短的方案。
[Prompt tokens: 8452, Completion tokens: 659]
---------- Summary ----------
Number of messages: 4
Finish reason: None
Total prompt tokens: 10967
Total completion tokens: 735
Duration: 46.22 seconds

Process finished with exit code 0