当前位置：首页 > news >正文

python mcp stdio模式

news 2026/7/4 19:49:21

server:

"""3.1 【stdio模式】mcp服务端开发"""
#导入mcp依赖包
from mcp.server.fastmcp import FastMCP
#创建mcp实例
mcp = FastMCP("Tool MCP Server")

@mcp.tool()
def add_tool(x:int,y:int):
"""
有两个数字相加的加法工具
:param x: 第一个数字
:param y: 第二个数字
:return: 两个数字的和
"""
return x+y

@mcp.tool()
def sub_tool(x:int,y:int):
"""
有两个数字相减的减法工具
:param x: 第一个数字
:param y: 第二个数字
:return: 两个数字的差
"""
return x-y

if __name__ == "__main__":
print(" MCP Server Start!")
#启动mcp服务:有两种协议，分别是stdio和tcp，stdio模式下，transport参数必须为stdio
mcp.run(transport="stdio")

client:
"""3.2 【stdio模式】mcp客户端开发—连接服务端"""
import asyncio
import json
from contextlib import AsyncExitStack

from mcp import StdioServerParameters, stdio_client, ClientSession
from openai import OpenAI

class MCPClient:

def __init__(self):
self.async_exit_stack = AsyncExitStack()
self.session = None
self.deepseek = OpenAI(
api_key="sk-304d80ba4865490283ec012fcdfa568a",
base_url="https://api.deepseek.com"
)

async def connect_to_server(self,server_path):
# 一、创建服务连接的参数
server_parameters = StdioServerParameters(
command="python",
args=[server_path],
env=None
)
# 二、创建stdio_client
client = stdio_client(server_parameters)
transport = await self.async_exit_stack.enter_async_context(client)
read_stream, write_stream = transport
# 三、创建会话client
client_session = ClientSession(read_stream, write_stream)
self.session = await self.async_exit_stack.enter_async_context(client_session)
# 四、初始化会话
await self.session.initialize()

async def execute(self,query:str):
# 一、获取server.py中的工具列表
response = await self.session.list_tools()
list_tools = response.tools
print("打印出获取的工具列表：",list_tools)
#二、创建function calling 格式(大模型使用）、
tools =[
{
"type":"function",
"function":{
"name":tool.name,
"description":tool.description,
"parameters":tool.inputSchema
}
} for tool in list_tools
]
# 三、创建messages,deepseek大模型的格式
messages = [
{
"role":"user",
"content":query
}
]
# 四、调用deepseek大模型
deepseek_response = self.deepseek.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tools
)
# 打印出大模型的决策结果
print("==== deepseek 响应持结果：",deepseek_response)
choice_result = deepseek_response.choices[0]
#第二次调用大模型的前置参数
messages.append(choice_result.message.model_dump())
tool_call = choice_result.message.tool_calls[0]

print(" tool_call:",tool_call)
print("大模型决策的最终结果,工具名称：",tool_call.function.name,",参数：",tool_call.function.arguments)
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
# 五、调用工具链
tool_result = await self.session.call_tool(
name = function_name,
arguments=arguments
)
print("==== 工具调用结果：",tool_result)
#最终的结果
result = tool_result.content[0].text
print("==== 最终的结果：",result)

# 六、使用大模型生成最终的结果，并且使用语言模型生成最终的结果
messages.append({
"role": "tool",
"content": tool_result.content[0].text,
"tool_call_id": tool_call.id
})
# 再次调用大模型
deepseek_response = self.deepseek.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tools,
)
# 获取最终的结果
result = deepseek_response.choices[0].message.content
print("==== 最终的结果：", result)

#关闭资源
async def aclose(self):
await self.async_exit_stack.aclose()

async def main():
client = MCPClient()
try:
await client.connect_to_server("server.py")
await client.execute("帮我计算一下2加3等于几？")
except Exception as e:
print(f"连接失败: {e}")
return
finally:
await client.aclose()

if __name__ == "__main__":
asyncio.run(main())

D:\Users\msi\miniconda3\python.exe D:\mcp-client\源码文档\源码文档\0722-01-01\07220101\mcp例子\client_stdio.py
[12/18/25 19:20:35] INFO Processing request of type server.py:674
ListToolsRequest
打印出获取的工具列表： [Tool(name='add_tool', title=None, description='\n 有两个数字相加的加法工具\n :param x: 第一个数字\n :param y: 第二个数字\n :return: 两个数字的和\n ', inputSchema={'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'add_toolArguments', 'type': 'object'}, outputSchema=None, icons=None, annotations=None, meta=None), Tool(name='sub_tool', title=None, description='\n 有两个数字相减的减法工具\n :param x: 第一个数字\n :param y: 第二个数字\n :return: 两个数字的差\n ', inputSchema={'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'sub_toolArguments', 'type': 'object'}, outputSchema=None, icons=None, annotations=None, meta=None)]
==== deepseek 响应持结果： ChatCompletion(id='7f5c79d0-5bba-46c2-86ad-d155997067da', choices=[Choice(finish_reason='tool_calls', index=0, logprobs=None, message=ChatCompletionMessage(content='我来帮您计算2加3等于几。', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=[ChatCompletionMessageFunctionToolCall(id='call_00_ovwMqIGcfUpVnZlRLJaM6xoE', function=Function(arguments='{"x": 2, "y": 3}', name='add_tool'), type='function', index=0)]))], created=1766056836, model='deepseek-chat', object='chat.completion', service_tier=None, system_fingerprint='fp_eaab8d114b_prod0820_fp8_kvcache', usage=CompletionUsage(completion_tokens=69, prompt_tokens=483, total_tokens=552, completion_tokens_details=None, prompt_tokens_details=PromptTokensDetails(audio_tokens=None, cached_tokens=0), prompt_cache_hit_tokens=0, prompt_cache_miss_tokens=483))
tool_call: ChatCompletionMessageFunctionToolCall(id='call_00_ovwMqIGcfUpVnZlRLJaM6xoE', function=Function(arguments='{"x": 2, "y": 3}', name='add_tool'), type='function', index=0)
大模型决策的最终结果,工具名称： add_tool ,参数： {"x": 2, "y": 3}
==== 工具调用结果： meta=None content=[TextContent(type='text', text='5', annotations=None, meta=None)] structuredContent=None isError=False
==== 最终的结果： 5
[12/18/25 19:20:39] INFO Processing request of type server.py:674
CallToolRequest
==== 最终的结果： 2加3等于5。

Process finished with exit code 0

查看全文

http://www.cnnetsun.cn/news/138560.html