一、什么是 AI Agent？为什么要搭？

简单说，AI Agent = 大模型 + 工具调用 + 自主规划。

普通 AI 聊天是"你问一句，它答一句"；而 Agent 是"你给一个目标，它自己拆解任务、调用工具、执行步骤、给你结果"。

举个例子：

• 普通 AI：你问"帮我查一下北京明天的天气"
• AI Agent：你问"帮我规划明天去北京出差的行程"，它会自己查天气 → 查高铁/航班 → 查酒店 → 生成完整方案

2026 年搭建 Agent 的技术栈已经非常成熟，核心就是 MCP 协议（Model Context Protocol）——由 Anthropic 提出、2025 年底移交 Linux Foundation 管理的开放标准，目前已成为 AI Agent 工具调用的行业事实标准。

本文用 Python 带你从零搭建一个可用的 Agent。

---

二、准备工作

2.1 环境要求

bash

# Python 3.10+
python --version

# 安装核心依赖
pip install openai  # 调用 LLM
pip install mcp     # MCP 协议客户端/服务端 SDK
pip install httpx   # HTTP 请求（工具调用用）

2.2 理解 MCP 架构

MCP 采用 Client-Server 架构：

┌─────────────────┐      MCP 协议       ┌──────────────────┐
│   AI Agent       │ ◄──────────────►   │   MCP Server     │
│  (MCP Client)    │                    │  (工具提供方)     │
│                  │                    │                  │
│   LLM + 规划器   │                    │  天气查询工具    │
│                  │                    │  文件搜索工具    │
│                  │                    │  数据库查询工具   │
└─────────────────┘                    └──────────────────┘

• MCP Server：提供具体工具的"服务员"（查天气、搜文件、调 API 等）
• MCP Client：Agent 本体，负责调度 LLM + 调用 Server 的工具

---

三、第一步：写一个 MCP Server（工具端）

我们先写一个最简单的 MCP Server，提供两个工具：查天气、算日期。

python

# weather_server.py
"""一个简单的 MCP Server：提供天气查询和日期计算工具"""
import asyncio
import json
from datetime import datetime, timedelta
from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent

# 创建 Server 实例
server = Server("my_tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """声明这个 Server 提供哪些工具"""
    return [
        Tool(
            name="get_weather",
            description="查询指定城市的当前天气",
            input_schema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 北京、上海"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="calculate_days",
            description="计算从今天到指定日期的天数差",
            input_schema={
                "type": "object",
                "properties": {
                    "target_date": {
                        "type": "string",
                        "description": "目标日期，格式 YYYY-MM-DD"
                    }
                },
                "required": ["target_date"]
            }
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """执行工具的具体逻辑"""
    if name == "get_weather":
        city = arguments["city"]
        # 这里用模拟数据，实际可以调真实天气 API
        weather_data = {
            "北京": {"temp": "28°C", "condition": "晴", "humidity": "45%"},
            "上海": {"temp": "26°C", "condition": "多云", "humidity": "60%"},
            "深圳": {"temp": "30°C", "condition": "阵雨", "humidity": "75%"},
        }
        result = weather_data.get(city, {"temp": "N/A", "condition": "未知"})
        return [TextContent(
            type="text",
            text=json.dumps(result, ensure_ascii=False)
        )]

    elif name == "calculate_days":
        target = datetime.strptime(arguments["target_date"], "%Y-%m-%d")
        today = datetime.now()
        delta = (target - today).days
        return [TextContent(
            type="text",
            text=json.dumps({
                "days_difference": delta,
                "today": today.strftime("%Y-%m-%d"),
                "target": arguments["target_date"]
            }, ensure_ascii=False)
        )]

    raise ValueError(f"未知工具: {name}")

async def main():
    """通过标准输入输出运行 Server"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

运行 Server：

bash

python weather_server.py

Server 会保持运行，等待 Client 通过标准输入输出连接。

---

四、第二步：写 Agent（Client 端）

这是 Agent 的核心——连接 LLM 和 MCP Server。

python

# agent.py
"""AI Agent 核心：连接 LLM + MCP 工具"""
import asyncio
import json
from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class SimpleAgent:
    """一个极简的 AI Agent"""

    def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1",
                 model: str = "gpt-4o"):
        self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
        self.model = model
        self.mcp_session = None
        self.tools = []          # MCP Server 提供的工具列表
        self.tool_handlers = {}  # 工具名称 -> 调用方法

    async def connect_mcp_server(self, command: str, args: list[str]):
        """连接 MCP Server"""
        server_params = StdioServerParameters(
            command=command,
            args=args
        )
        read_stream, write_stream = await stdio_client(server_params).__aenter__()
        self.mcp_session = await ClientSession(read_stream, write_stream).__aenter__()
        await self.mcp_session.initialize()

        # 获取 Server 提供的工具列表
        response = await self.mcp_session.list_tools()
        self.tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.input_schema
            }
        } for tool in response.tools]

        # 保存工具调用方法
        self.tool_handlers = {
            tool.name: tool for tool in response.tools
        }

    async def run(self, user_message: str, max_turns: int = 10) -> str:
        """运行 Agent：用户输入 -> LLM 思考 -> 调用工具 -> 返回结果"""
        messages = [{"role": "user", "content": user_message}]

        for turn in range(max_turns):
            # 1. 调用 LLM
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=self.tools if self.tools else None,
                tool_choice="auto"
            )

            message = response.choices[0].message
            messages.append(message)

            # 2. 检查 LLM 是否要调用工具
            if not message.tool_calls:
                # LLM 直接返回了最终答案
                return message.content

            # 3. 执行工具调用
            for tool_call in message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                print(f"\n  🔧 调用工具: {tool_name}({tool_args})")

                # 通过 MCP 协议调用 Server 上的工具
                result = await self.mcp_session.call_tool(tool_name, tool_args)
                result_text = result.content[0].text if result.content else ""

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

                print(f"  ✅ 工具返回: {result_text[:100]}...")

        return "Agent 已达到最大执行轮次。"


async def main():
    """使用示例"""
    # 创建 Agent
    agent = SimpleAgent(
        api_key="your-api-key-here",  # 替换为你的 API Key
        base_url="https://api.openai.com/v1",  # 可换成其他兼容接口
        model="gpt-4o"
    )

    # 连接我们刚写的 MCP Server
    await agent.connect_mcp_server(
        command="python",
        args=["weather_server.py"]
    )

    # 运行 Agent
    result = await agent.run("我后天要去北京出差，帮我查一下北京天气怎么样？")
    print(f"\n=== Agent 回答 ===\n{result}")


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

运行 Agent：

bash

python agent.py

---

五、效果演示

运行后，Agent 会自动拆解"后天"是什么日期 → 调用日期计算工具 → 调用天气查询工具 → 汇总回答：

  调用工具: calculate_days({"target_date": "2026-06-07"})
  工具返回: {"days_difference": 2, "today": "2026-06-05", "target": "2026-06-07"}

  调用工具: get_weather({"city": "北京"})
  工具返回: {"temp": "28°C", "condition": "晴", "humidity": "45%"}

=== Agent 回答 ===
后天（6月7日）北京天气晴好，气温28°C，湿度45%，非常适合出行！
建议带一件薄外套，早晚可能略有温差。祝你出差顺利！

---

六、进阶：挂载多个 MCP Server

一个 Agent 可以同时连接多个 MCP Server，组合能力：

async def connect_multiple_servers(agent):
    """连接多个 MCP Server，组合不同能力"""
    servers = [
        ("python", ["weather_server.py"]),          # 天气
        ("python", ["search_server.py"]),           # 搜索
        ("python", ["file_server.py"]),             # 文件操作
        ("node", ["database-server/index.js"]),     # 数据库（用 Node.js 写的也行）
    ]

    all_tools = []
    all_handlers = {}

    for command, args in servers:
        try:
            server_params = StdioServerParameters(command=command, args=args)
            read, write = await stdio_client(server_params).__aenter__()
            session = await ClientSession(read, write).__aenter__()
            await session.initialize()

            response = await session.list_tools()
            for tool in response.tools:
                all_tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.input_schema
                    }
                })
                all_handlers[tool.name] = (session, tool)

            print(f"  ✅ 已连接: {' '.join(args)}")
        except Exception as e:
            print(f"  ❌ 连接失败 {args}: {e}")

    agent.tools = all_tools
    agent.tool_handlers = all_handlers

---

七、部署建议与避坑

7.1 选型建议

场景	推荐方案
个人小工具	Python + MCP SDK，本文方案即可
企业内部	Spring AI + MCP（Java 生态）
生产级多 Agent	LangGraph + LangChain
鸿蒙应用	Agent Framework Kit（FunctionComponent）

7.2 常见坑

1. 工具描述要写好：LLM 靠 description 字段判断何时调用哪个工具，描述越清楚，Agent 越聪明
2. 超时处理：工具调用要设超时，避免某个工具卡死整个 Agent
3. 安全隔离：MCP Server 应该有独立的权限控制，不能直接暴露给用户
4. 幂等设计：工具最好幂等，多次调用同一参数结果一致
5. 日志要全：记录每次工具调用的入参和出参，方便排查

7.3 在鸿蒙上做 Agent

如果你用的是 HarmonyOS 开发，官方提供了 Agent Framework Kit（API 11+）：

typescript

// HarmonyOS 6+ 内嵌 Agent 到 ArkTS 页面
import { FunctionComponent, AgentController } from '@kit.AgentKit';

@Entry
@Component
struct MyAgentPage {
  private controller: AgentController = new AgentController();

  aboutToAppear() {
    // 先检查环境是否支持
    this.controller.checkAvailability().then((ok: boolean) => {
      if (!ok) {
        console.error('当前环境不支持 Agent');
      }
    });
  }

  build() {
    Column() {
      FunctionComponent({
        agentId: 'my_custom_agent',
        controller: this.controller,
        onError: (err) => {
          // 区分错误类型，给出用户友好提示
          if (err.code === 'NETWORK_ERROR') {
            // 提示检查网络
          }
        }
      })
    }
  }
}

---

八、总结

搭建一个 AI Agent 其实就三步：

1. 写 MCP Server：把你想给 AI 用的工具封装成标准接口
2. 写 Agent Client：连接 LLM + MCP Server，让 AI 能调用你的工具
3. 组合升级：挂载多个 Server，让 Agent 能力越来越强

本文所有代码都可以直接运行，完整代码不到 100 行。建议你先跑通 demo，再根据实际需求扩展工具。

相关资源：

• MCP 协议官方文档
• MCP Python SDK
• HarmonyOS Agent Framework Kit 文档

---

本文发布于 2026年6月5日。MCP 协议版本以 Linux Foundation 官方为准，HarmonyOS 部分基于 API 11+。