玩转MCP

mcp火了好一阵了，最近一直在大院干活儿，好不容易抽出时间，赶紧来学习学习。
官方文档， mcp广场可以搜索自己需要的server服务，都是些saas服务，其实就是把各家的server下载到本地然后调用。官方分stdio和sse两种交互模式。我这里都用代码自定义，方便项目应用，代码放到仓库了。需要使用claude，vscode等的可以自行搜索使用方法，我这里不涉及。

0.环境

conda create -n mcp python==3.12
conda activate mcp
pip install mcp openai

1.自定义stdio交互

1.1.server

服务端包括三部分：创建mcp客户端，然后写自己的tool都带上@mcp.tool()注解，最后run（这里确定使用stdio方式还是sse方式）

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my_server", version="1.0.0")@mcp.tool()
def use_my_camera()-> None:passif __name__ == "__main__":mcp.run(transport='stdio')

仓库代码里我写的四个tool，打开本地相机，向本地写入文件，打开spotify并播放某首歌曲，给我一个leetcode题目，写的比较糙，慎用，需要的可以去下载官网的。
这里我在调用spotify的时候记得要去https://developer.spotify.com/dashboard创建自己的开发者账户，拿到自己的client_id和client_secret，这里redirect url最好要写http://127.0.0.1:8888/callback/

1.2.client

客户端包括三部分：连接大模型(我用ollama，qwen2.5)，连接mcp服务器(这里stdio和sse有区别)，处理用户请求（将大模型和mcp服务建立连接，这里需要根据自己的模型效果改改逻辑）。

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI
import json
import sys
class MCPClient:def __init__(self):# 初始化会话和客户端对象self.session: Optional[ClientSession] = None  # 使用会话管理和 API 客户端进行初始化self.exit_stack = AsyncExitStack()  # 使用AsyncExitStack管理资源self.openai_api_key = "ollama" self.base_url = "http://192.168.3.36:11434/v1"  #Ollama 服务器的 IP 地址self.model = "qwen2.5:latest"  # Ollama 模型名称self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)async def connect_to_server(self, server_script_path: str):"""连接到MCP服务器Args:server_script_path (str): 服务器脚本路径（.py 或.js）"""if not server_script_path.endswith(('.py', '.js')):raise ValueError("服务器脚本必须是.py 或.js 文件")command = "python" if server_script_path.endswith('.py') else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))# 初始化会话并列出可用工具await self.session.initialize()response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器, 可用工具:", [tool.name for tool in tools])async def process_query(self, query: str) -> str:"""处理查询，使用Claude和可用工具Args:query (str): 用户输入的查询Returns:str: 处理后的响应内容"""# 构造消息对象messages = [{"role": "user","content": query}]# 获取可用工具列表response = await self.session.list_tools()# 将工具列表转换为需要的格式available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema}}for tool in response.tools]# 第一次调用Claude API# print("进入》》》》》》》process_query")response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)print("response",response)# 获取API响应的内容content = response.choices[0].messagefinal_text = []assistant_message_content = []if content.tool_calls:# 获取工具调用信息tool_call = content.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 调用工具result = await self.session.call_tool(tool_name, tool_args)final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")assistant_message_content.append(content)print(f"\n\n[正在用工具 {tool_name}, 参数为 {tool_args}]\n\n")# 修正消息列表更新逻辑messages.append({"role": "assistant","content": "","tool_calls": [tool_call]})messages.append({"role": "tool","content": result.content[0].text if result.content[0].text else "","name": tool_name,"tool_call_id": tool_call.id})response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)print("response",response)# 获取API响应的内容content = response.choices[0].messagefinal_text.append(content.content)# 如果没有调用工具，直接返回响应内容else:final_text.append(content.content)assistant_message_content.append(content)return "\n".join(final_text)async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP 客户端已启动！")print("请输入您的查询或输入 'quit' 退出。")while True:try:query = input("user input:").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)print("\n" + response)except Exception as e:print(f"\n错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv) < 2:print("用法: python client.py <path_to_server_script>")sys.exit(1)client = MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

1.3.效果

运行

python client.py server.py

输入命令进行测试，我的命令（写一封500字告白信到我本地，给我播放一首好日子，我想练一个简单的二分法，打开本地相机）。
还可以使用网页端测试效果

pip install mcp[cli]
mcp dev server.py

2.自定义sse交互

sse交互与stdio差别很小，这里只写区别

2.1.server

这里初始化mcp的时候给一个端口号，运行的时候使用sse模式，剩下工具的定义都不变。

mcp = FastMCP("my_server", version="1.0.0", port=8000)
@mcp.tool()
if __name__ == "__main__":mcp.run(transport='sse')

2.2.client

client这里就是连接mcp服务的时候与stdio不同，然后在main里面调用的时候传递的就是server启动的url地址

async def connect_to_server(self, url: str):"""连接到基于 HTTP SSE 的 MCP 工具服务器"""sse_transport = await self.exit_stack.enter_async_context(sse_client(url))self.session = await self.exit_stack.enter_async_context(ClientSession(*sse_transport))await self.session.initialize()response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器, 可用工具:", [tool.name for tool in tools])
async def main():client = MCPClient()try:await client.connect_to_server('http://0.0.0.0:8000/sse')await client.chat_loop()finally:await client.cleanup()

2.3.效果

启动服务端python server.py
启动客户端python client.py
网页端测试mcp dev server.py

3.使用官方文件

这里我测试了一个操作键鼠，截屏等操作的ScreenPilot，首先git代码，进入环境。
安装依赖包pip install -r requirements.txt
他的代码是stdio的交互（我没找到sse的），所以client我就用第一个stdio里面的，server就用官方的main文件，所以运行python C:\Users\robot\Desktop\mcp\stdio\client.py C:\Users\robot\Desktop\mcp\ScreenPilot\main.py就可以调用这个服务。