AI王炸：MCP服务端客户端的完整实现

显示全部楼层

概述

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;font-weight: 300;letter-spacing: normal;orphans: 2;text-align: start;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">试想一下，如果要想在现有应用上构建，让AI读取引用我们功能和数据，该怎么办，比如询问某个城市的天气，我们希望AI能调用天气函数返回相应结果，这时MCP(Model Context Protocol)就可以派上用场了，它相当于我们电脑的USB-C接口，提供了一个标准方式让AI模型连接不同的引用和工具。我们可以建一个MUP Server来处理这类业务，比如市面上已有各类MUP Server，比较典型的高德地图MCP，除此之外，还有旅行交通的AirbnbMCPServer，提供房源问询，版本控制的gitlab-mr-mcp，工具类mcp-openai，开发类mcp-server-and-gw等，更多工具可查看：https://github.com/punkpeye/awesome-mcp-servers/blob/main/README-zh.md#%E6%9C%8D%E5%8A%A1%E5%99%A8%E5%AE%9E%E7%8E%B0

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;font-weight: 300;letter-spacing: normal;orphans: 2;text-align: start;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">MCP 服务器的职能变得非常容易理解：即遵循 MCP 协议来暴露其可提供的 Resources、Tools 或 Prompts：

Resources：结构化数据（如文件、API 响应）
Tools：可执行函数（如查询数据库、发送邮件）
Prompts：预定义的交互模板

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;font-weight: 300;letter-spacing: normal;orphans: 2;text-align: start;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">MCP与Function Caling的区别：

开发一个MCP Server

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;font-weight: 300;letter-spacing: normal;orphans: 2;text-align: start;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">使用Python为例。

安装uv

windows：powershell-ExecutionPolicyByPass-c"irmhttps://astral.sh/uv/install.ps1|iex"linux/mac：curl-LsSfhttps://astral.sh/uv/install.sh|sh

安装依赖

pipinstallmcppipinstallmcp[cli]pipinstallhttpx==0.27

编写代码：非常简单，将普通的py函数添加一个注解即可。

importosfrommcp.server.fastmcpimportFastMCPmcp=FastMCP("债券服务Demo")@mcp.tool()deffilterByRate(a:str)->list:"""根据债券评级的条件筛选债券"""print('债券评级过滤条件：',a)return["债券A","债券B","债券C"]@mcp.tool()deffilterByType(t:str)->list:"""根据债券类型的条件筛选债券"""print('债券类型过滤条件：',t)return["债券A","债券C","债券D"]@mcp.tool()deffilterByBidRange(a1:float,a2:float)->list:"""根据债券bid收益率区间筛选债券"""print('债券类型过滤条件：',a1,a2)return["债券A2","债券C2","债券D2"]@mcp.tool()deffilterResult(**kwargs)->list:"""获取所有符合条件的债券结果"""print('filterResult：',kwargs)return["债券A111","债券C222","债券D3333"]if__name__=="__main__":#mcp.run(transport='stdio')mcp.run(transport='sse')

启动服务：mcp dev.\mcp\hello.py

客户端集成

工具集成

ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-style: normal;font-variant-ligatures: normal;font-variant-caps: normal;font-weight: 300;letter-spacing: normal;orphans: 2;text-align: start;text-indent: 0px;text-transform: none;widows: 2;word-spacing: 0px;-webkit-text-stroke-width: 0px;text-decoration-thickness: initial;text-decoration-style: initial;text-decoration-color: initial;">以CherryStudio为例，其他类型，添加一个：MCPServer,问答时选中MCP Server即可.

代码集成

程序集成：编写一个客户端程序即可，我们以本地Ollama部署的qwen为例，代码如下：

client_mcp.pyimportasyncioimportjsonimportsysimporttimefromtypingimportOptionalfromcontextlibimportAsyncExitStackfrommcp.client.sseimportsse_clientfrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientfromopenaiimportAsyncOpenAIclassMCPClient:def__init__(self):#Initializesessionandclientobjectsself.session:Optional[ClientSession]=Noneself.exit_stack=AsyncExitStack()self.client=AsyncOpenAI(api_key="test",base_url="http://ollama地址:11434/v1")asyncdefconnect_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])asyncdefconnect_to_sse_server(self,server_url:str):"""ConnecttoanMCPserverArgs:server_script_pathathtotheserverscript(.pyor.js)"""self._streams_context=sse_client(url=server_url)streams=awaitself._streams_context.__aenter__()self._session_context=ClientSession(*streams)self.session=awaitself._session_context.__aenter__()awaitself.session.initialize()#Listavailabletoolsresponse=awaitself.session.list_tools()tools=response.toolsprint("\nConnectedtoserverwithtools:",[tool.namefortoolintools])asyncdefprocess_query(self,query:str)->str:"""使用LLM和MCP服务器提供的工具处理查询"""messages=[{"role":"user","content":query}]response=awaitself.session.list_tools()available_tools=[{"type":"function","function":{"name":tool.name,"description":tool.description,"parameters":tool.inputSchema}}fortoolinresponse.tools]#初始化LLMAPI调用response=awaitself.client.chat.completions.create(model="qwen2.5:14b",messages=messages,tools=available_tools#将工具列表传递给LLM)final_text=[]message=response.choices[0].messageprint(response.choices[0])final_text.append(message.contentor"")#处理响应并处理工具调用ifmessage.tool_calls:#处理每个工具调用fortool_callinmessage.tool_calls:tool_name=tool_call.function.nametool_args=json.loads(tool_call.function.arguments)#执行工具调用start_time=time.time()result=awaitself.session.call_tool(tool_name,tool_args)end_time=time.time()print(f"Tool{tool_name}took{end_time-start_time}secondstoexecute")final_text.append(f"[Callingtool{tool_name}withargs{tool_args}]")#将工具调用和结果添加到消息历史messages.append({"role":"assistant","tool_calls":[{"id":tool_call.id,"type":"function","function":{"name":tool_name,"arguments":json.dumps(tool_args)}}]})messages.append({"role":"tool","tool_call_id":tool_call.id,"content":str(result.content)})#将工具调用的结果交给LLMresponse=awaitself.client.chat.completions.create(model="qwen2.5:14b",messages=messages,tools=available_tools)message=response.choices[0].messageifmessage.content:final_text.append(message.content)return"\n".join(final_text)asyncdefchat_loop(self):"""Runaninteractivechatloop"""print("\nMCPClientStarted!")print("Typeyourqueriesor'quit'toexit.")whileTrue:try:query=input("\nQuery:").strip()ifquery.lower()=='quit':breakresponse=awaitself.process_query(query)print("\n"+response)exceptExceptionase:print(f"\nError:{str(e)}")asyncdefcleanup(self):"""Cleanupresources"""awaitself.exit_stack.aclose()main.pyimportrequestsimportjsonimporthttpximportasynciofromclient_mcpimportMCPClientimportsysasyncdefmain():url_server_mcp='http://localhost:8000/sse'client=MCPClient()try:#根据MCPServer传输协议进行选择awaitclient.connect_to_sse_server(url_server_mcp)awaitclient.chat_loop()finally:awaitclient.cleanup()if__name__=='__main__':loop=asyncio.get_event_loop()loop.run_until_complete(main())