code/05-agents/mcp_client.py

"""连接 mcp_server.py：先列出工具、直接调用一次；再把 MCP 工具交给大模型，让它自己决定怎么用。

在 AI-Course/code/05-agents 目录下运行：python mcp_client.py
"""
import asyncio
import json
import os
import sys
from pathlib import Path

from mcp import Client, StdioServerParameters
from openai import OpenAI

# 告诉客户端怎么启动服务器：用当前的 Python 解释器运行 mcp_server.py
SERVER = StdioServerParameters(command=sys.executable, args=[str(Path(__file__).parent / "mcp_server.py")])
llm = OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"))
MODEL = os.environ.get("LLM_MODEL", "deepseek-flash")


def text_of(result):
    """MCP 工具的返回是一组内容块，把其中的文字拼起来。"""
    return "\n".join(block.text for block in result.content if getattr(block, "text", None))


async def main():
    async with Client(SERVER) as mcp:
        # 1. 服务器有哪些工具？
        tools = (await mcp.list_tools()).tools
        print("服务器提供的工具：")
        for t in tools:
            print(f"  {t.name}：{t.description}")
            print(f"    参数：{json.dumps(t.input_schema['properties'], ensure_ascii=False)}")

        # 2. 不经过大模型，直接调用一次
        result = await mcp.call_tool("grep_docs", {"keyword": "http2=True"})
        print("\n直接调用 grep_docs('http2=True')：")
        print(text_of(result))

        # 3. 把 MCP 的工具说明转成 OpenAI 接口的格式，交给大模型
        schemas = [{"type": "function", "function": {
            "name": t.name, "description": t.description, "parameters": t.input_schema}} for t in tools]
        messages = [{"role": "user", "content": "httpx 怎么开启 HTTP/2？需要先装什么？注明文档出处。"}]
        for step in range(1, 7):
            msg = llm.chat.completions.create(model=MODEL, messages=messages, tools=schemas,
                                              extra_body={"thinking": {"type": "disabled"}}).choices[0].message
            if not msg.tool_calls:
                print(f"\n大模型的回答：\n{msg.content}")
                break
            messages.append(msg.model_dump(exclude_none=True))
            for call in msg.tool_calls:
                # 大模型要调用的工具，转发给 MCP 服务器去执行
                result = await mcp.call_tool(call.function.name, json.loads(call.function.arguments))
                print(f"[第 {step} 步] 通过 MCP 调用 {call.function.name}({call.function.arguments})")
                messages.append({"role": "tool", "tool_call_id": call.id, "content": text_of(result)})


asyncio.run(main())