"""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="sse")
"""3.2 【stdio模式】mcp客户端开发—连接服务端"""
import asyncio
import json
from contextlib import AsyncExitStack
from mcp import StdioServerParameters, stdio_client, ClientSession
from mcp.client.sse import sse_client
from openai import OpenAI
import time
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,url:str):
# 一 、创建stdio_client
client = sse_client(url=url)
transport = await self.async_exit_stack.enter_async_context(client)
print('----------------------transport---------------------------')
print(transport)
print('----------------------transport---------------------------')
read_stream, write_stream = transport
print('----------------read_stream, write_stream------------------')
print(read_stream,write_stream)
print('----------------read_stream, write_stream------------------')
# 二、创建会话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服务端中的工具列表
response = await self.session.list_tools()
list_tools = response.tools
print("responce",response)
print("打印出获取的工具列表：",list_tools)
#二、创建function calling 格式(大模型使用）、
tools =[
{
"type":"function",
"function":{
"name":tool.name,
"description":tool.description,
"parameters":tool.inputSchema
}
} for tool in list_tools
]
print('------------tools-------------------')
print(tools)
print('------------tools-------------------')
# 三、 创建messages,deepseek大模型的格式
messages = [
{
"role":"user",
"content":query
}
]
print ('--------------------message0-----------------')
print(messages)
print('--------------------message0-----------------')
# 四、调用deepseek大模型
print('---------------message1--------------------')
print(messages)
print('---------------message1--------------------')
deepseek_response = self.deepseek.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tools
)
# 打印出大模型的决策结果
print("==== deepseek 响应持结果：",deepseek_response)
choice_result = deepseek_response.choices[0]
print('--------------choice_result---------------------')
print(choice_result)
print('--------------choice_result---------------------')
#第二次调用大模型的前置参数
messages.append(choice_result.message.model_dump())
print('----------------messages2----------------')
print(messages)
print('----------------messages2----------------')
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)
print('------function_name,arguments-------------------')
print(function_name,arguments)
print('------function_name,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
})
print('----------------messages3----------------')
print(messages)
print('----------------messages3----------------')
# 再次调用大模型
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()
await client.connect_to_server("http://127.0.0.1:8000/sse")
await client.execute("帮我计算一下2加3等于几？")
await client.aclose()
if __name__ == "__main__":
asyncio.run(main())
