stdio传输通过标准输入和输出流进行通信。这在本地集成和命令行工具中尤其有用。
使用 stdio 的场景包括:
(Claude就是基于这个能力实现电脑控制的)
服务器发送事件(SSE)使服务器能够通过HTTP POST请求向客户端进行流式传输,用于客户端与服务器之间的通信。
使用SSE的情况包括:
本文以介绍如何构建MCP Server为例,主要内容为基于Python SDK自定义SSE传输类型的MCP Server,并向外部应用提供tools的能力。
Python SDK的github地址为 https://github.com/modelcontextprotocol/python-sdk ,其提供了一系列的demo实例,例如可以通过fastmcp创建:
# server.py
frommcp.server.fastmcpimportFastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
defadd(a:int, b:int) ->int:
"""Add two numbers"""
returna + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
defget_greeting(name:str) ->str:
"""Get a personalized greeting"""
returnf"Hello,{name}!"然后直接启动:
mcpdevserver.py
或者,也可通过lowlevel的Server创建
importanyio
importclick
importhttpx
importmcp.typesastypes
frommcp.server.lowlevelimportServer
# 此处代码省略,完整代码参考:https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/servers/simple-tool/mcp_simple_tool/server.py
# ...
@click.command()
@click.option("--port", default=8000,help="Port to listen on for SSE")
@click.option(
"--transport",
type=click.Choice(["stdio","sse"]),
default="stdio",
help="Transport type",
)
defmain(port:int, transport:str) ->int:
app = Server("mcp-website-fetcher")
@app.call_tool()
asyncdeffetch_tool(
name:str, arguments:dict
) ->list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
ifname !="fetch":
raiseValueError(f"Unknown tool:{name}")
if"url"notinarguments:
raiseValueError("Missing required argument 'url'")
returnawaitfetch_website(arguments["url"])
@app.list_tools()
asyncdeflist_tools() ->list[types.Tool]:
return[
types.Tool(
name="fetch",
description="Fetches a website and returns its content",
inputSchema={
"type":"object",
"required": ["url"],
"properties": {
"url": {
"type":"string",
"description":"URL to fetch",
}
},
},
)
]
iftransport =="sse":
frommcp.server.sseimportSseServerTransport
fromstarlette.applicationsimportStarlette
fromstarlette.routingimportMount, Route
sse = SseServerTransport("/messages/")
asyncdefhandle_sse(request):
asyncwithsse.connect_sse(
request.scope, request.receive, request._send
)asstreams:
awaitapp.run(
streams[0], streams[1], app.create_initialization_options()
)
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
importuvicorn
uvicorn.run(starlette_app, host="0.0.0.0", port=port)
else:
frommcp.server.stdioimportstdio_server
asyncdefarun():
asyncwithstdio_server()asstreams:
awaitapp.run(
streams[0], streams[1], app.create_initialization_options()
)
anyio.run(arun)
return0然后启动该main函数。(具体介绍可参考代码中链接)
当你完成Server构建之后,即可使用MCP Inspector工具进行调试。如果你使用fastMCP进行构建,那么mcp dev server.py命令默认会启动MCP Inspector。而如果使用lowlevel的Server创建,则默认不会,那么此时可通过额外启动MCP Inspector来进行调试。可执行如下命令:
npx@modelcontextprotocol/inspector然后打开本地5173端口,即可得到:
本文使用到的mcp server代码地址:
https://github.com/xiaozhch5/chatdata-mcp-server
修改URL地址,连接MCP Server:现在我们已经定义好了MCP Server,且通过sse的方式将tools暴露出来。接着,便可以通过此sse将这些tools接近来。
McpTransporttransport=
newHttpMcpTransport.Builder()
.sseUrl("http://127.0.0.1:8099/sse")
.logRequests(true)// if you want to see the traffic in the log
.logResponses(true)
.build();
McpClientmcpClient=newDefaultMcpClient.Builder().transport(transport).build();
ToolProvidertoolProvider=McpToolProvider.builder().mcpClients(List.of(mcpClient)).build();
ChatAgentAssistantbot=
AiServices.builder(ChatAgentAssistant.class)
.chatLanguageModel(openAiCompatibleModel.getChatModel("gpt-3.5-turbo"))
.toolProvider(toolProvider)
.build();
Stringres=bot.chat("""
帮我查一下广州的天气
""");
System.out.println(res);执行上述代码,即可得到:通过构建基于MCP协议的服务器并向外部系统暴露工具服务,我们不仅实现了与不同语言的大模型应用无缝对接,还提供了高效的调试路径及最佳实践。从快速搭建服务器到详细地使用MNP Inspector进行功能验证,再到在实际项目中测试和优化,整个过程帮助开发者高效集成和管理大模型组件,从而为用户提供更丰富、更友好的交互体验。
本文使用的MCP Server相关代码地址(纯用cursor写的):
https://github.com/xiaozhch5/chatdata-mcp-server
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |