用MCP Server为你的大模型应用插上翅膀

玩大模型的同学应该对MCP（Model Context Protocol）都不陌生。去年Anthropic发布该大模型协议之后，就引起了广泛关注，因为有了它，大模型就可以插上翅膀。

MCP是一个开源协议，它规定了应用程序如何向LLM提供上下文。可以把MCP想象成AI应用的Type C数据线。就像Type C数据线为将您的设备连接到各种外部设备和配件提供了标准化方式一样，MCP也为将AI模型连接到不同的数据源和工具提供了标准化方法。

MCP基本架构

借助python SDK，我们可以定义Resources、 Tools、 Roots、Sampling 以及 Prompt。

两种标准传输实现

MCP提供了Standard Input/Output (stdio)和Server-Sent Events (SSE)两种传输实现。

Standard Input/Output (stdio)

stdio传输通过标准输入和输出流进行通信。这在本地集成和命令行工具中尤其有用。

使用 stdio 的场景包括：

• • 构建命令行工具
• • 实现本地集成
• • 需要简单的进程通信
• • 与 shell 脚本进行交互

（Claude就是基于这个能力实现电脑控制的）

Server-Sent Events (SSE)

服务器发送事件（SSE）使服务器能够通过HTTP POST请求向客户端进行流式传输，用于客户端与服务器之间的通信。

使用SSE的情况包括：

• • 仅需要从服务器到客户端的单向流式传输
• • 在受限网络环境下工作
• • 实现简单的更新功能

本文以介绍如何构建MCP Server为例，主要内容为基于Python SDK自定义SSE传输类型的MCP Server，并向外部应用提供tools的能力。

MCP Server创建

Python SDK的github地址为 https://github.com/modelcontextprotocol/python-sdk ，其提供了一系列的demo实例，例如可以通过fastmcp创建：

# server.py
from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")

# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

然后直接启动：

mcp dev server.py

或者，也可通过lowlevel的Server创建

import anyio
import click
import httpx
import mcp.types as types
from mcp.server.lowlevel import Server

# 此处代码省略，完整代码参考：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",
)
def main(port: int, transport: str) -> int:
    app = Server("mcp-website-fetcher")

    @app.call_tool()
    async def fetch_tool(
        name: str, arguments: dict
    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
        if name != "fetch":
            raise ValueError(f"Unknown tool: {name}")
        if "url" not in arguments:
            raise ValueError("Missing required argument 'url'")
        return await fetch_website(arguments["url"])

    @app.list_tools()
    async def list_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",
                        }
                    },
                },
            )
        ]

    if transport == "sse":
        from mcp.server.sse import SseServerTransport
        from starlette.applications import Starlette
        from starlette.routing import Mount, Route

        sse = SseServerTransport("/messages/")

        async def handle_sse(request):
            async with sse.connect_sse(
                request.scope, request.receive, request._send
            ) as streams:
                await app.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),
            ],
        )

        import uvicorn

        uvicorn.run(starlette_app, host="0.0.0.0", port=port)
    else:
        from mcp.server.stdio import stdio_server

        async def arun():
            async with stdio_server() as streams:
                await app.run(
                    streams[0], streams[1], app.create_initialization_options()
                )

        anyio.run(arun)

    return 0

然后启动该main函数。（具体介绍可参考代码中链接）

当你完成Server构建之后，即可使用MCP Inspector工具进行调试。如果你使用fastMCP进行构建，那么mcp dev server.py命令默认会启动MCP Inspector。而如果使用lowlevel的Server创建，则默认不会，那么此时可通过额外启动MCP Inspector来进行调试。可执行如下命令：

npx @modelcontextprotocol/inspector

本文使用到的mcp server代码地址：

https://github.com/xiaozhch5/chatdata-mcp-server

现在我们已经定义好了MCP Server，且通过sse的方式将tools暴露出来。接着，便可以通过此sse将这些tools接近来。

langchain4j接入MCP Server实例

McpTransport transport =  
    new HttpMcpTransport.Builder()  
        .sseUrl("http://127.0.0.1:8099/sse")  
        .logRequests(true) // if you want to see the traffic in the log  
        .logResponses(true)  
        .build();  
McpClient mcpClient = new DefaultMcpClient.Builder().transport(transport).build();  
  
ToolProvider toolProvider = McpToolProvider.builder().mcpClients(List.of(mcpClient)).build();  
  
ChatAgentAssistant bot =  
    AiServices.builder(ChatAgentAssistant.class)  
        .chatLanguageModel(openAiCompatibleModel.getChatModel("gpt-3.5-turbo"))  
        .toolProvider(toolProvider)  
        .build();  
  
String res = bot.chat("""  
            帮我查一下广州的天气
            """);  
System.out.println(res);

总结

通过构建基于MCP协议的服务器并向外部系统暴露工具服务，我们不仅实现了与不同语言的大模型应用无缝对接，还提供了高效的调试路径及最佳实践。从快速搭建服务器到详细地使用MNP Inspector进行功能验证，再到在实际项目中测试和优化，整个过程帮助开发者高效集成和管理大模型组件，从而为用户提供更丰富、更友好的交互体验。

题外话

MCP Server地址

本文使用的MCP Server相关代码地址（纯用cursor写的）：

https://github.com/xiaozhch5/chatdata-mcp-server

Browser Use使用