AI Agent 工具调用：Function Call、MCP 与 A2A 技术演进

import json
from openai import OpenAI

client = OpenAI(api_key="你的 API_KEY")

# 真实的工具函数 (由你的代码执行)
def get_weather(city: str) -> str:
    weather_db = {"北京": "晴天 25°C", "上海": "小雨 22°C", "吉林市": "大雪 -20°C"}
    return weather_db.get(city, "未知城市天气")

# 注册工具
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的实时天气状况",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"}
            },
            "required": ["city"]
        }
    }
}]

messages = [{"role": "user", "content": "吉林市现在天气怎么样？"}]

# 第一次调用：让大模型决策
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

response_message = response.choices[0].message

if response_message.tool_calls:
    tool_call = response_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    
    print(f"🤖 大模型决策：调用 '{function_name}'，参数 {function_args}")
    
    # ⭐️ 关键分界线：此处由宿主代码真正执行
    real_result = get_weather(**function_args)
    print(f"💻 宿主代码执行结果：{real_result}")
    
    # 回传结果，生成最终回答
    messages.append(response_message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": real_result
    })
    
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    print("🗣️ 最终回答:", final_response.choices[0].message.content)

🤖 大模型决策：调用 'get_weather'，参数 {'city': '吉林市'}
💻 宿主代码执行结果：大雪 -20°C
🗣️ 最终回答：吉林市现在是大雪天气，气温低至零下 20°C，请注意保暖!

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气状况"""
    weather_db = {"北京": "晴天 25°C", "上海": "小雨 22°C", "吉林市": "大雪 -20°C"}
    return weather_db.get(city, "未知城市天气")

model = ChatOpenAI(model="gpt-4o").bind_tools([get_weather])
messages = [HumanMessage(content="吉林市现在天气怎么样？")]
response = model.invoke(messages)

print(response.content)  # 直接获得最终回答

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def main():
    client = MultiServerMCPClient({
        "math": {  # 数学计算 MCP 服务器
            "transport": "stdio",
            "command": "python",
            "args": ["/path/to/math_server.py"],
        },
        "weather": {  # 天气查询 MCP 服务器
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        },
    })
    
    tools = await client.get_tools()  # 统一加载所有工具
    agent = create_agent("claude-sonnet-4-6", tools)
    
    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "计算 (3+5)x12"}]}
    )

curl --request POST \
  --url https://api.example.com/a2a/{assistant_id} \
  --header 'Content-Type: application/json' \
  --data '{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "message/send",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"kind": "text", "text": "请帮我查询明天的航班"}],
        "messageId": "msg-1"
      }
    }
  }'