基于python的MCP使用简单教程

MCP是一种新型的agent通信协议,其突出贡献是统一了各模型厂商的function call各不相同的局面。这里基于windows11 和python 3.11环境对MCP的使用作简单示例(模型调用的是类openai API的格式,现在各大厂商一般都支持)。 接下来让我们5分钟上手mcp:

一、新建环境

安装annaconda(ps:官方uv安装一堆bug,这里就用conda替代了)

conda create -n mcp python==3.11

conda activate mcp

pip install pqi

pqi use tuna

pip install httpx fastmcp asyncio python-dotenv openai mcp

新建mcp-test文件夹,并cd到其中。

二、mcp-test文件夹中放入这三个文件:

1、客户端文件

client.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
import os
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv() # load environment variables from .env

api_key = os.environ["DS_API_KEY"]
base_url = os.environ["DS_API_BASE"]


# What are the weather alert in California
class MCPClient:
def __init__(self):
# Initialize session and client objects
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()

self.openai = OpenAI(api_key=api_key, base_url=base_url)

async def connect_to_server(self, server_script_path: str):
"""Connect to an MCP server

Args:
server_script_path: Path to the server script (.py or .js)
"""
is_python = server_script_path.endswith(".py")
is_js = server_script_path.endswith(".js")
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")

command = "python" if is_python 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_transport
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.stdio, self.write)
)

await self.session.initialize()

# List available tools
response = await self.session.list_tools()
tools = response.tools
print(
"\nConnected to server with tools:",
[[tool.name, tool.description, tool.inputSchema] for tool in tools],
)

async def process_query(self, query: str) -> str:
"""Process a query using Claude and available tools"""
messages = [{"role": "user", "content": query}]

response = await self.session.list_tools()

available_tools = []

for tool in response.tools:
tool_schema = getattr(
tool,
"inputSchema",
{"type": "object", "properties": {}, "required": []},
)

openai_tool = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool_schema,
},
}
available_tools.append(openai_tool)

# Initial Claude API call
model_response = self.openai.chat.completions.create(
model="ep-20250221170925-gg4xx",
max_tokens=1000,
messages=messages,
tools=available_tools,
)

# Process response and handle tool calls
tool_results = []
final_text = []

messages.append(model_response.choices[0].message.model_dump())
print('工具调用指令:',messages[-1])

if model_response.choices[0].message.tool_calls:
# breakpoint()
tool_call = model_response.choices[0].message.tool_calls[0]
tool_args = json.loads(tool_call.function.arguments)

tool_name = tool_call.function.name
result = await self.session.call_tool(tool_name, tool_args)
print(f"工具名称: {tool_name} 调用参数: {tool_args} \n工具返回结果: {result}")
tool_results.append({"call": tool_name, "result": result})
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

messages.append(
{
"role": "tool",
"content": f"{result}",
"tool_call_id": tool_call.id,
}
)

# Get next response from Claude
response = self.openai.chat.completions.create(
model="ep-20250221170925-gg4xx",
max_tokens=1000,
messages=messages,
)

messages.append(response.choices[0].message.model_dump())
print('模型回复结果:',messages[-1])

return messages[-1]["content"]

async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")

while True:
try:
query = input("\nQuery: ").strip()

if query.lower() == "quit":
break

response = await self.process_query(query)
print("\n" + response)

except Exception as e:
print(f"\nError: {str(e)}")

async def cleanup(self):
"""Clean up resources"""
await self.exit_stack.aclose()


async def main():
if len(sys.argv) < 2:
print("Usage: 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__":
import sys

asyncio.run(main())

2、服务端文件

tools.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化 FastMCP 服务器
mcp = FastMCP("tools")

@mcp.tool()
async def count_letters(word: str, letter: str) -> str:
"""统计单词中特定字母出现的次数。

参数:
word: 要分析的单词
letter: 要统计的字母(单个字符)
"""
# 确保 letter 是单个字符
if len(letter) != 1:
return "请提供单个字母进行统计"

# 将输入转换为小写以忽略大小写差异
word = word.lower()
letter = letter.lower()

# 统计字母出现次数
count = word.count(letter)

return f"单词 '{word}' 中有 {count} 个 '{letter}'"

@mcp.tool()
async def compare_numbers(num1: float, num2: float) -> str:
"""比较两个数字的大小。

参数:
num1: 第一个数字
num2: 第二个数字
"""
if num1 > num2:
return f"{num1}{num2} 大"
elif num1 < num2:
return f"{num1}{num2} 小"
else:
return f"{num1}{num2} 相等"

@mcp.tool()
async def get_current_date_time() -> str:
"""获取当前时间和日期。"""
import datetime
now = datetime.datetime.now()
return f'当前时间为:{now.strftime("%Y-%m-%d %H:%M:%S")}'

if __name__ == "__main__":
# 初始化并运行服务器
mcp.run(transport='stdio')

3、配置文件

.env

1
2
DS_API_KEY=xxx  #根据你的api key填写
DS_API_BASE=https://xxx/api/v1 #根据你请求的具体地址填写

三、运行带服务的mcp客户端

在mcp-test路径下执行:

1
python client.py tools.py

服务开始运行,输入你的指令,如

query:当前时间是?

回车后显示:

image-20250322214756793

好了,以上你已经入门MCP了,更加详细的内容可以参考官方博客: https://modelcontextprotocol.info/docs/introduction/


基于python的MCP使用简单教程
https://linxkon.github.io/基于python的MCP使用简单教程.html
作者
linxkon
发布于
2025年3月22日
许可协议