MCP 协议客户端与服务端python demo
mcp server 端:
# -*- coding: utf-8 -*-
'''
@Date 2025-09-15
'''"""
MCP server服务端
需要安装mcp 包 "mcp[cli]"
pip install "mcp[cli]"
或者 uv add mcp[cli]
"""from mcp.server.fastmcp import FastMCP
from pathlib import Path
from typing import List
mcp = FastMCP("testmcp",stateless_http=False)@mcp.tool()
async def get_sum(a:int,b:int) -> int:"""Calculate the sum of two integersArgs:a intb int"""c = a+breturn c@mcp.tool()
async def get_files(path:str) -> List[str]:"""Obtain all the files in the path directoryArgs:path str query the path directory"""dir_path = Path(path)files = [file.name for file in dir_path.iterdir() if file.is_file()]return filesif __name__ == "__main__":# Initialize and run the server#transport = stdio|sse|streamable-httpmcp.run(transport='streamable-http')
mcp client端:
# -*- coding: utf-8 -*-
'''
@Date 2025-09-16
'''"""
MCP client客户端
"""import requests
import json
import randomclass StreamableHttpClient(object):"""mcp streamable-http 方式mcp协议 官网：https://modelcontextprotocol.comhttps://modelcontextprotocol.io/specification/2025-06-18/basic/lifecycle连接MCP 服务的时候必须先 "initialize"和"notifications/initialized"然后再进行功能操作"""def __init__(self,url):""""""self.url = urlself.session = requests.Session()self.random_id = random.randint(1,10)self.mcp_session_id = ""self.init_notification = Falsecode,resp,result = self.initialize()if code == 200:code,resp,result = self.notification_initialized()if code == 202:self.init_notification = Trueelse:print("notification_initialized error")def initialize(self):""""""client_name = "python-streamable-client"client_version = "1.0.0"payload = {"jsonrpc": "2.0","method": 'initialize',"id": self.random_id,"params": {"protocolVersion": "2025-06-18","capabilities": {"tools": {},"resources": {},"prompts": {},"sampling": {},"elicitation":{},"roots": {'listChanged':True}},"clientInfo": {"name": client_name,"version": client_version,'title':client_name}}}headers = {'Content-Type': 'application/json','Accept': 'application/json, text/event-stream'}self.random_id += 1self.session.headers.update(headers)resp = self.session.post(self.url,json=payload,stream=False)self.mcp_session_id = resp.headers.get('mcp-session-id',"")return self._pro_response(resp)def notification_initialized(self):"""notification 通知从客户端发送到服务器，或者从服务器发送到客户端，作为单向消息。接收方不得发送响应注意不需要 payload 中不需要参数"id"status_code == 202 不存在text和json()"""payload = {"jsonrpc": "2.0","method": 'notifications/initialized'}headers = {'Content-Type': 'application/json','Accept': 'application/json, text/event-stream'}headers['Mcp-Session-Id'] = self.mcp_session_idself.random_id += 1self.session.headers.update(headers)resp = self.session.post(self.url,json=payload,stream=False)return self._pro_response(resp)def tools_list(self):""""""if not self.init_notification:print("tools_list request error")return 111,None,{}payload = {"jsonrpc": "2.0","method": 'tools/list',"id": self.random_id,"params":{}}headers = {'Content-Type': 'application/json','Accept': 'application/json, text/event-stream','Mcp-Session-Id':self.mcp_session_id,}self.random_id += 1self.session.headers.update(headers)resp = self.session.post(self.url,json=payload,stream=True)return self._pro_response(resp)def tools_call(self,name:str,args:dict):""""""if not self.init_notification:print("tools_list request error")return 222,None,{}payload = {"jsonrpc": "2.0","method": 'tools/call',"id":self.random_id,"params":{'name':name,"arguments":args}}self.random_id += 1headers = {'Content-Type': 'application/json','Accept': 'application/json, text/event-stream','Mcp-Session-Id':self.mcp_session_id,}self.session.headers.update(headers)resp = self.session.post(self.url,json=payload,stream=True)return self._pro_response(resp)def _pro_response(self,resp):"""notification_initialized status_code == 202 不存在text和json()"""if resp.status_code != 200:return resp.status_code,resp,{}ct = resp.headers.get("content-type","").lower()if 'text/event-stream' in ct:result = json.loads(resp.text.strip().split("data: ")[1])elif 'application/json' in ct:result = resp.json()else:result = {}return resp.status_code,resp,resulturl = "http://127.0.0.1:8000/mcp"
c = StreamableHttpClient(url)
code,resp,result = c.tools_list()
print('tools_list',code,result)
code,resp,result = c.tools_call('get_sum',{"a":2,"b":3})
print('tools_call',code,result)