手搓Manus？MCP 原理解析与MCP Client实践 - 链载Ai

MCP（Model Context Protocol）是由Anthropic于2024年底提出并开源的一种协议，旨在为AI系统（如AI编程助手、Agent等）提供安全、标准化的数据访问方式。它采用客户端-服务器架构，使AI工具（如Claude Desktop、IDE插件等）能够通过MCP客户端与MCP服务端交互，访问本地或远程数据源。

协议层（Protocol Layer）

传输层（Transport Layer）

类似于三次握手，MCP客户端与MCP服务端初始化建立连接会进行以下步骤：

初始化（Initialization）

消息交换（Message Exchange）

终止（Termination）

基础概念介绍完毕，接下来进行实践，我希望能实现一个自己的 agent，让 AI 不仅能和我交流，还能帮我干活。换一句话就是

myAgent is a general AI agent that turns your thoughts into actions. It excels at various tasks in work and life, getting everything done while you rest.

[手动doge][手动doge]

如图，要实现一个这样的效果，实现一个 myAgent，启动时，MCP Client建立与 MCP 服务端的连接，此时 MCP Server 上的能力或者可调用的工具会注册进来，让 Client 感知到这个MCP服务能够干啥。

当用户与 Agent 进行交互时，Agent 会让 MCP Client 将用户的输入发送给 AI，让 AI 解析用户意图，一并发送的还有注册在 Client 上的能力集合。

我写了一个搜索助手的 MCP Server ，能力集的数据是这样的，可以看到目前只有一个 function，get_offers，可以看到工具里有它的名字，能力描述，需要的字段（包含字段类型，字段描述）。

AI 会理解用户意图，决定是否用自然语言回答用户，或者选择合适的工具，告诉 client，帮我调用它。

当输入你好时，传给 AI 的 message 是这样的，这里系统预设了 AI 的一个身份，用于更好的完成特定领域的任务。

此时 client 接收到 AI 的消息后，会解析数据，当没有工具要调用时

可以看到，AI 返回是带有角色信息的，然后本次并没有识别到需要调用工具的地方，因此直接返回给用户就好，当然，在工程应用时，可以进行额外的逻辑处理。

第一次AI 交互

可以看到，AI 识别到了用户的意图，是要寻找一些手表，并自动的选择了一个工具进行调用，根据工具使用说明，决定了选择的工具应该输入什么入参。(这里和模型很相关，是一个重要的节点，识别用户意图并决定要调用工具，有时识别的并不准确，或者返回的结构不是标准可解析的，这时就触发不了工具的调用，还会引入一些辣鸡信息，可能的解决方案是换效果更好的模型，或者用提示词来约束模型返回，或者系统自己增加鲁棒性，提升成功率）

下面举一个效果不好的例子，大家如果知道有其他解决方法欢迎留言。

Client 调用 MCP Server

client 接收到AI 的消息后，发现要调用工具，并且也有了工具所需的参数，就会与通过协议与 MCP Server 进行通信，告诉 MCP Server 该使用 get_offers 能力了，并且期待 MCP Server 将结果返回回来：

获取 MCP Server 数据

MCP Server 不负众望，将结果返回了，可以看看返回的格式是什么样的：

第二次 AI 交互

这里给了它 10 个品，但是只总结了两个品，可能适合我之前的输入帮我找一些手表有关，看来AI 也会偷懒?。

上面的交互过程，其实可以化简，如果在工程应用上，调用的 MCP Server 是一个预期内的结构化的结果或者触发某个任务时，可以不必进行二次 AI 调用。如上面的例子中，MCP Server 是一个搜索助手，内部发起调用的是搜索的接口，并进行结构化返回。此时在 AI 识别到用户意图并告诉 Client 该调用什么工具时，与 AI 的交互就可以结束了，由系统接管决定应该返回给用户什么，不必再将结果给到 AI 进行润色总结。

给到 AI 进行润色总结的好处是可以根据用户的输入，再结合工具获取的数据，更智能友好的返回给用户信息，这一点可以在工程应用时，进行衡量取舍。

将MCP Server 的结果交给 AI，在需要进行多轮交互场景是有必要的，根据 MCP Server的结果，进行分析及决策，动态调整要使用的工具，可以将一个复杂的任务交给 AI , 它会自己拆解成小任务，然后自动完成。

例如，让Al agent拆解抽象任务，并自己主动与系统进行多轮交互，完成任务场景。

后续文章我们会详细介绍具体实现方法，讲讲如何让Al在我的电脑上玩起贪吃蛇？

importasyncioimportjsonimportosimporttracebackfromtypingimportOptionalfromcontextlibimportAsyncExitStackfrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientfromopenaiimportOpenAIfromdotenvimportload_dotenvload_dotenv()#loadenvironmentvariablesfrom.envclassMCPClient:def__init__(self):#Initializesessionandclientobjectsself.session:Optional[ClientSession]=Noneself.exit_stack=AsyncExitStack()self.client=OpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL"))self.model=os.getenv("OPENAI_MODEL")self.messages=[{"role":"system","content":"Youareaversatileassistantcapableofansweringquestions,completingtasks,andintelligentlyinvokingspecializedtoolstodeliveroptimalresults."}]self.available_tools=[]@staticmethoddefconvert_custom_object(obj):"""将自定义对象转换为字典"""ifhasattr(obj,"__dict__"):#如果对象有__dict__属性，直接使用returnobj.__dict__elifisinstance(obj,(list,tuple)):#如果是列表或元组，递归处理return[MCPClient.convert_custom_object(item)foriteminobj]elifisinstance(obj,dict):#如果是字典，递归处理值return{key:MCPClient.convert_custom_object(value)forkey,valueinobj.items()}else:#其他类型（如字符串、数字等）直接返回returnobjasyncdefconnect_to_server(self,server_script_path:str):"""ConnecttoanMCPserverArgs:server_script_pathathtotheserverscript(.pyor.js)"""is_python=server_script_path.endswith('.py')is_js=server_script_path.endswith('.js')ifnot(is_pythonoris_js):raiseValueError("Serverscriptmustbea.pyor.jsfile")command="python"ifis_pythonelse"node"server_params=StdioServerParameters(command=command,args=[server_script_path],env=None)stdio_transport=awaitself.exit_stack.enter_async_context(stdio_client(server_params))self.stdio,self.write=stdio_transportself.session=awaitself.exit_stack.enter_async_context(ClientSession(self.stdio,self.write))awaitself.session.initialize()#Listavailabletoolsresponse=awaitself.session.list_tools()tools=response.toolsprint("\nConnectedtoserverwithtools:",[tool.namefortoolintools])asyncdefprocess_query(self,query:str)->str:"""rocessaquerywithmulti-turntoolcallingsupport"""#Adduserquerytomessagehistoryself.messages.append({"role":"user","content":query})#Getavailabletoolsifnotalreadysetifnotself.available_tools:response=awaitself.session.list_tools()self.available_tools=[{"type":"function","function":{"name":tool.name,"description":tool.description,"parameters":tool.inputSchema}}fortoolinresponse.tools]current_response=self.client.chat.completions.create(model=self.model,messages=self.messages,tools=self.available_tools,stream=False)#Printinitialresponseifexistsifcurrent_response.choices[0].message.content:print("\n?AI:",current_response.choices[0].message.content)#直到下一次交互AI没有选择调用工具时退出循环whilecurrent_response.choices[0].message.tool_calls:#AI一次交互中可能会调用多个工具fortool_callincurrent_response.choices[0].message.tool_calls:tool_name=tool_call.function.nametry:tool_args=json.loads(tool_call.function.arguments)exceptjson.JSONDecodeError:tool_args={}print(f"\n?调用工具{tool_name}")print(f"?参数:{tool_args}")#Executetoolcallresult=awaitself.session.call_tool(tool_name,tool_args)print(f"\n工具结果:{result}")#AddAImessageandtoolresulttohistoryself.messages.append(current_response.choices[0].message)self.messages.append({"role":"tool","tool_call_id":tool_call.id,"content":json.dumps(result.content)})#Getnextresponsecurrent_response=self.client.chat.completions.create(model=self.model,messages=self.messages,tools=self.available_tools,stream=False)#Addfinalresponsetohistoryself.messages.append(current_response.choices[0].message)returncurrent_response.choices[0].message.contentor""asyncdefchat_loop(self):"""Runaninteractivechatloop"""print("\nMCPClientStarted!")print("Typeyourqueriesor'quit'toexit.")whileTrue:try:query=input("\nCommend:").strip()ifquery.lower()=='quit':breakresponse=awaitself.process_query(query)print("\n?AI:"+response)exceptExceptionase:print(f"\nErroroccurs:{e}")traceback.print_exc()asyncdefcleanup(self):"""Cleanupresources"""awaitself.exit_stack.aclose()asyncdefmain():iflen(sys.argv)<2:print("Usage:pythonclient.py<path_to_server_script>")sys.exit(1)client=MCPClient()try:awaitclient.connect_to_server(sys.argv[1])awaitclient.chat_loop()finally:awaitclient.cleanup()if__name__=="__main__":importsysasyncio.run(main())

/***MCP客户端实现**提供与MCP服务器的连接、工具调用和聊天交互功能**主要功能：*1.连接Python或JavaScript实现的MCP服务器*2.获取服务器提供的工具列表*3.通过OpenAIAPI处理用户查询*4.自动处理工具调用链*5.提供交互式命令行界面**使用说明：*1.确保设置OPENAI_API_KEY环境变量*2.通过命令行参数指定MCP服务器脚本路径*3.启动后输入查询或'quit'退出**依赖：*-@modelcontextprotocol/sdk:MCP协议SDK*-openai:OpenAIAPI客户端*-dotenv:环境变量加载*/import{Client}from"@modelcontextprotocol/sdk/client/index.js";import{StdioClientTransport}from"@modelcontextprotocol/sdk/client/stdio.js";importOpenAIfrom"openai";importtype{ChatCompletionMessageParam}from"openai/resources/chat/completions";importtype{Tool}from"@modelcontextprotocol/sdk/types.js";import*asdotenvfrom"dotenv";import*asreadlinefrom'readline';//加载环境变量配置dotenv.config();/***MCP客户端类，封装与MCP服务器的交互逻辑*/classMCPClient{privateopenai:OpenAI;//OpenAIAPI客户端实例privateclient:Client;//MCP协议客户端实例privatemessages:ChatCompletionMessageParam[]=[{role:"system",content:"Youareaversatileassistantcapableofansweringquestions,completingtasks,andintelligentlyinvokingspecializedtoolstodeliveroptimalresults."},];//聊天消息历史记录，用于维护对话上下文privateavailableTools:any[]=[];//服务器提供的可用工具列表，格式化为OpenAI工具格式/***构造函数，初始化OpenAI和MCP客户端**@throws{Error}如果OPENAI_API_KEY环境变量未设置**初始化过程：*1.检查必要的环境变量*2.创建OpenAI客户端实例*3.创建MCP客户端实例*4.初始化消息历史记录*/constructor(){if(!process.env.OPENAI_API_KEY){thrownewError("OPENAI_API_KEY环境变量未设置");}this.openai=newOpenAI({apiKey:process.env.OPENAI_API_KEY,baseURL:process.env.OPENAI_BASE_URL,});this.client=newClient({name:"my-mcp-client",version:"1.0.0",},);}/***连接到MCP服务器**@param{string}serverScriptPath-服务器脚本路径(.py或.js)*@returns{Promise<void>}连接成功时解析*@throws{Error}如果服务器脚本不是.py或.js文件，或连接失败**连接过程：*1.检查脚本文件扩展名*2.根据扩展名决定使用python或node执行*3.通过stdio建立连接*4.获取服务器工具列表并转换为OpenAI工具格式**注意事项：*-服务器脚本必须具有可执行权限*-连接成功后会自动获取工具列表*/asyncconnectToServer(serverScriptPath:string){constisPython=serverScriptPath.endsWith('.py');constisJs=serverScriptPath.endsWith('.js');if(!isPython&&!isJs){thrownewError("Serverscriptmustbea.pyor.jsfile");}constcommand=isPython?"python":"node";consttransport=newStdioClientTransport({command,args:[serverScriptPath],});awaitthis.client.connect(transport);//获取并转换可用工具列表consttools=(awaitthis.client.listTools()).toolsasunknownasTool[];this.availableTools=tools.map(tool=>({type:"function"asconst,function:{name:tool.nameasstring,description:tool.descriptionasstring,parameters:{type:"object",properties:tool.inputSchema.propertiesasRecord<string,unknown>,required:tool.inputSchema.requiredasstring[],},}}));console.log("\n已连接到服务器，可用工具:",tools.map(tool=>tool.name));}/***处理工具调用链**@param{OpenAI.Chat.Completions.ChatCompletion}response-初始OpenAI响应，包含工具调用*@param{ChatCompletionMessageParam[]}messages-当前消息历史记录*@returns{Promise<OpenAI.Chat.Completions.ChatCompletion>}最终OpenAI响应**处理流程：*1.检查响应中是否包含工具调用*2.循环处理所有工具调用*3.解析每个工具调用的参数*4.执行工具调用*5.将工具结果添加到消息历史*6.获取下一个OpenAI响应**错误处理：*-参数解析失败时使用空对象继续执行*-工具调用失败会抛出异常**注意事项：*-此方法会修改传入的messages数组*-可能多次调用OpenAIAPI*/privateasynctoolCalls(response:OpenAI.Chat.Completions.ChatCompletion,messages:ChatCompletionMessageParam[]){letcurrentResponse=response;//直到下一次交互AI没有选择调用工具时退出循环while(currentResponse.choices[0].message.tool_calls){if(currentResponse.choices[0].message.content){console.log("\n?AI:tool_calls",JSON.stringify(currentResponse.choices[0].message));}//AI一次交互中可能会调用多个工具for(consttoolCallofcurrentResponse.choices[0].message.tool_calls){consttoolName=toolCall.function.name;constrawArgs=toolCall.function.arguments;lettoolArgs;try{console.log(`rawArgsis=====${rawArgs}`)toolArgs="{}"==JSON.parse(rawArgs)?{}:JSON.parse(rawArgs);if(typeoftoolArgs==="string"){toolArgs=JSON.parse(toolArgs);}}catch(error){console.error('⚠️参数解析失败，使用空对象替代');toolArgs={};}console.log(`\n?调用工具${toolName}`);console.log(`?参数:`,toolArgs);//调用工具获取结果constresult=awaitthis.client.callTool({name:toolName,arguments:toolArgs});console.log(`\nresultis${JSON.stringify(result)}`);//添加AI的响应和工具调用结果到消息历史//console.log(`?currentResponse.choices[0].message:`,currentResponse.choices[0].message);messages.push(currentResponse.choices[0].message);messages.push({role:"tool",tool_call_id:toolCall.id,content:JSON.stringify(result.content),}asChatCompletionMessageParam);}//console.log(`?messages:`,messages);//获取下一个响应currentResponse=awaitthis.openai.chat.completions.create({model:process.env.OPENAI_MODELasstring,messages:messages,tools:this.availableTools,});}returncurrentResponse;}/***处理用户查询**@param{string}query-用户输入的查询字符串*@returns{Promise<string>}AI生成的响应内容**处理流程：*1.将用户查询添加到消息历史*2.调用OpenAIAPI获取初始响应*3.如果有工具调用，处理工具调用链*4.返回最终响应内容**错误处理：*-OpenAIAPI调用失败会抛出异常*-工具调用链中的错误会被捕获并记录**注意事项：*-此方法会更新内部消息历史*-可能触发多个工具调用*/asyncprocessQuery(query:string)romise<string>{//添加用户查询到消息历史this.messages.push({role:"user",content:query,});//初始OpenAIAPI调用letresponse=awaitthis.openai.chat.completions.create({model:process.env.OPENAI_MODELasstring,messages:this.messages,tools:this.availableTools,});//打印初始响应if(response.choices[0].message.content){console.log("\n?AI:",response.choices[0].message);}//处理工具调用链if(response.choices[0].message.tool_calls){response=awaitthis.toolCalls(response,this.messages);}//更新消息历史this.messages.push(response.choices[0].message);returnresponse.choices[0].message.content||"";}/***启动交互式聊天循环**@returns{Promise<void>}当用户退出时解析**功能：*1.持续接收用户输入*2.处理用户查询*3.显示AI响应*4.输入'quit'退出**实现细节：*-使用readline模块实现交互式输入输出*-循环处理直到用户输入退出命令*-捕获并显示处理过程中的错误**注意事项：*-此方法是阻塞调用，会一直运行直到用户退出*-确保在调用前已连接服务器*/asyncchatLoop(){console.log("\nMCPClientStarted!");console.log("Typeyourqueriesor'quit'toexit.");constrl=readline.createInterface({input:process.stdin,output:process.stdout,});while(true){constquery=awaitnewPromise<string>((resolve)=>{rl.question("\nQuery:",resolve);});if(query.toLowerCase()==='quit'){break;}try{constresponse=awaitthis.processQuery(query);console.log("\n"+response);}catch(e){console.error("\nError:",einstanceofError?e.message:String(e));}}rl.close();}/***清理资源**@returns{Promise<void>}资源清理完成后解析**关闭以下资源：*1.MCP客户端连接*2.任何打开的句柄**最佳实践：*-应在程序退出前调用*-建议在finally块中调用以确保执行**注意事项：*-多次调用是安全的*-清理后实例不可再用*/asynccleanup(){if(this.client){awaitthis.client.close();}}}/***主函数**程序入口点，执行流程：*1.检查命令行参数*2.创建MCP客户端实例*3.连接到指定服务器脚本*4.启动交互式聊天循环*5.退出时清理资源**@throws{Error}如果缺少命令行参数或连接失败**使用示例：*```bash*nodeindex.js/path/to/server.js*```**退出码：*-0:正常退出*-1:参数错误或运行时错误*/asyncfunctionmain(){if(process.argv.length<3){console.log("Usage:nodedist/index.js<path_to_server_script>");process.exit(1);}constclient=newMCPClient();try{awaitclient.connectToServer(process.argv[2]);awaitclient.chatLoop();}finally{awaitclient.cleanup();}}main().catch((error)=>{console.error("Error:",error);process.exit(1);});

#!/usr/bin/env nodeimport{Server}from'@modelcontextprotocol/sdk/server/index.js';import{StdioServerTransport}from'@modelcontextprotocol/sdk/server/stdio.js';import{CallToolRequestSchema,ErrorCode,ListToolsRequestSchema,McpError,}from'@modelcontextprotocol/sdk/types.js';importaxiosfrom'axios';
interfaceProduct{id:string;name:string;supplier:string; 省略...}
interfaceOffer{productId:string;title:string;companyName:string; 省略...}
interfaceApiResponse{ret:string[];encode:string;code:number;traceId:string;msg:string;time:number;data: { offers:Offer[]; resultCount:string; totalCount:number; };}
classProductOffersServer{privateserver:Server;privatebaseUrl ='换成你要调用的 url';
constructor() { this.server=newServer(   {   name:'search-assistant-server',   version:'0.1.0',   },   {   capabilities: {    tools: {},    },   }  );
 this.setupToolHandlers();  // Error handling this.server.onerror=(error) =>console.error('[MCP Error]', error);  process.on('SIGINT',async() => {  awaitthis.server.close();   process.exit(0);  }); }
privateasyncfetchOffers(keywords?:string, pageSize?:number)romise<roduct[]> { try{  constparams:Record<string,any> = {};  if(keywords) params.keywords= keywords;  if(pageSize) params.pageSize= pageSize;  constresponse =awaitaxios.get<ApiResponse>(this.baseUrl, { params });    returnresponse.data.data.offers.map(offer=>({   id: offer.productId,   name: offer.title,   supplier: offer.companyName    省略...   }));  }catch(error) {  if(axios.isAxiosError(error)) {   thrownewMcpError(    ErrorCode.InternalError,    `Failed to fetch offers{error.message}`    );   }  throwerror;  } }
privatesetupToolHandlers() { this.server.setRequestHandler(ListToolsRequestSchema,async() => ({  tools: [    {    name:'get_offers',    description:'Get product offers from API',    inputSchema: {     type:'object',     properties: {      keywords: {       type:'string',       description:'Keywords to search for products',       default:''       },      pageSize: {       type:'number',       description:'Number of items per page',       minimum:1,       maximum:100,       default:10       }      }     }    }   ]  }));
 this.server.setRequestHandler(CallToolRequestSchema,async(request) => {  if(request.params.name!=='get_offers') {   thrownewMcpError(    ErrorCode.MethodNotFound,    `Unknown tool{request.params.name}`    );   }
  constargs = request.params.argumentsas{ keywords?:string; pageSize?:number};
  try{   constproducts =awaitthis.fetchOffers(args.keywords, args.pageSize);   return{    content: [      {      type:'text',      text:JSON.stringify({       products: products,       totalCount: products.length       },null,2)      }     ]    };   }catch(error) {   if(errorinstanceofMcpError) {    throwerror;    }   thrownewMcpError(    ErrorCode.InternalError,    `Failed to fetch offers{error}`    );   }  }); }
asyncrun() { consttransport =newStdioServerTransport(); awaitthis.server.connect(transport); console.error('roduct Offers MCP server running on stdio'); }}
constserver =newProductOffersServer();server.run().catch(console.error);