精通模型上下文协议（MCP）：基于 Azure OpenAI 构建多服务器 MCP 实践

1
2
3

from langchain_openai import AzureChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, FileResponse
import uvicorn

app = FastAPI()

@app.post("/ask")
async def ask(request: Request):
    data = await request.json()
    message = data.get("message", "")
    result = await run_agent(message)
    return JSONResponse(content={"result": result})

@app.get("/")
async def root():
    return FileResponse("frontend.html", media_type="text/html")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool(name="Addition", description="Adds two integers")
def add(a: int, b: int) -> int:
    return a + b

if __name__ == "__main__":
    mcp.run(transport="stdio")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
  "servers": {
    "mcp_foundry_server": {
      "type": "stdio",
      "command": "uvx",
      "args": [
        "--prerelease=allow",
        "--from",
        "git+https://github.com/azure-ai-foundry/mcp-foundry.git",
        "run-azure-ai-foundry-mcp",
        "--envFile",
        "${workspaceFolder}/.env"
      ]
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import json, os

MCP_CLIENT_CONFIG = {}

with open(os.path.join(".vscode", "mcp.json"), "r") as f:
    mcp_config = json.load(f)
for name, cfg in mcp_config["servers"].items():
    args = [arg.replace("${workspaceFolder}", os.getcwd()) for arg in cfg["args"]]
    MCP_CLIENT_CONFIG[name] = {
        "command": cfg["command"],
        "args": args,
        "transport": cfg["type"]
    }
client = MultiServerMCPClient(MCP_CLIENT_CONFIG)

1
2
3
4

async def get_tools():
    async with client:
        tools = client.get_tools()
        return [{"name": t.name, "description": getattr(t, "description", "")} for t in tools]

1
2
3
4
5
6

async def run_agent(message):
    async with client:
        tools = client.get_tools()
        agent = create_react_agent(model, tools)
        agent_response = await agent.ainvoke({"messages": message})
        return agent_response["messages"][3].content

1
2
3
4
5
6
7
8

@app.get("/tools")
async def get_tools():
    async with client:
        tools = client.get_tools()
        return {"tools": [
            {"name": t.name, "description": getattr(t, "description", "")}
            for t in tools
        ]}