MCP协议详解：一文读懂跨时代的模型上下文协议

{uri: string; // Unique identifier for the resourcename: string;// Human-readable namedescription?: string;// Optional descriptionmimeType?: string; // Optional MIME type}

{name: string;// Unique identifier for the promptdescription?: string;// Human-readable descriptionarguments?: [// Optional list of arguments{name: string;// Argument identifierdescription?: string;// Argument descriptionrequired?: boolean;// Whether argument is required}]}

{name: string;// Unique identifier for the tooldescription?: string;// Human-readable descriptioninputSchema: { // JSON Schema for the tool's parameterstype: "object",properties: { ... }// Tool-specific parameters}}

{messages: [{role: "user" | "assistant",content: {type: "text" | "image",
// For text:text?: string,
// For images:data?: string, // base64 encodedmimeType?: string}}],modelPreferences?: {hints?: [{name?: string// Suggested model name/family}],costPriority?: number, // 0-1, importance of minimizing costspeedPriority?: number,// 0-1, importance of low latencyintelligencePriority?: number// 0-1, importance of capabilities},systemPrompt?: string,includeContext?: "none" | "thisServer" | "allServers",temperature?: number,maxTokens: number,stopSequences?: string[],metadata?: Record<string, unknown>}

from typing import Anyfrom mcp.server.fastmcp import FastMCP# Initialize FastMCP servermcp = FastMCP("Path of Exile 2 hotfix")target_url = "https://www.pathofexile.com/forum/view-forum/2212"

async def poe2_hotfix(url: str):headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'}response = requests.get(url, headers=headers)async with httpx.AsyncClient() as client:try:response = await client.get(url, headers=headers, timeout=30.0)soup = BeautifulSoup(response.text, 'html.parser')# 查找包含帖子内容的表格table = soup.find('table')result_text = ""if table:for row in table.find_all('tr'):cells = row.find_all('td')if cells:for cell in cells:result_text += cell.get_text(strip=True) + '\n'result_text += '-' * 50 + '\n' # 分隔线else:print('未找到表格元素')return result_textexcept Exception:return None

@mcp.tool()async def find_poe2_hotfix() -> str:hotfix_data = await poe2_hotfix(target_url)if not hotfix_data:return "Unable to find any hotfix in office"return hotfix_data

if __name__ == "__main__":# Initialize and run the server mcp.run(transport='stdio')

pip install mcpmcp dev server.py

uv run client.py poe2hotfix.py

a. 从 MCP 服务器动态发现工具b. 工具描述自动包含在系统提示中c. 工具执行通过标准化 MCP 协议处理

a. 收到用户输入b. 输入与可用工具的上下文一起发送到 LLMc. LLM 响应解析：如果是工具调用 → 执行工具并返回结果如果是直接响应 → 返回给用户d. 工具结果发送回 LLM 进行解释e. 最终响应呈现给用户

all_tools = []for server in self.servers:tools = await server.list_tools()all_tools.extend(tools)

def format_for_llm(self) -> str:"""Format tool information for LLM.
Returns:A formatted string describing the tool."""args_desc = []if "properties" in self.input_schema:for param_name, param_info in self.input_schema["properties"].items():arg_desc = (f"- {param_name}: {param_info.get('description', 'No description')}")if param_name in self.input_schema.get("required", []):arg_desc += " (required)"args_desc.append(arg_desc)
return f"""Tool: {self.name}Description: {self.description}Arguments:{chr(10).join(args_desc)}"""

system_message = ("You are a helpful assistant with access to these tools:\n\n"f"{tools_description}\n""Choose the appropriate tool based on the user's question. ""If no tool is needed, reply directly.\n\n""IMPORTANT: When you need to use a tool, you must ONLY respond with ""the exact JSON object format below, nothing else:\n""{\n"'"tool": "tool-name",\n''"arguments": {\n''"argument-name": "value"\n'"}\n""}\n\n""After receiving a tool's response:\n""1. Transform the raw data into a natural, conversational response\n""2. Keep responses concise but informative\n""3. Focus on the most relevant information\n""4. Use appropriate context from the user's question\n""5. Avoid simply repeating the raw data\n\n""Please use only the tools that are explicitly defined above.")