手把手教你从 0 到 1 构建 MCP Server & Client【附教程】

MCP Server 是实现模型上下文协议（MCP）的服务器，旨在为 AI 模型提供一个标准化接口，连接外部数据源和工具，例如文件系统、数据库或 API。

相比之下，在MCP出现前，AI调用工具基本上是通过Function Call 完成的，通过Function Call 调取相关Function 或 API 调用相关工具，AI 模型根据用户提示生成函数调用指令，这些指令随后由系统执行，例如查询天气或管理文件。但是存在两个问题：

1. 不同的大模型厂商 Function Call 的格式不一致

2. 大量的 api 工具的输入和输出格式不一致，封装管理起来繁琐不方便

而 MCP 相当于是一个统一的 USB-C，不仅统一了不同大模型厂商的 Function Call 格式，也对相关工具的封装进行了统一。 

今天 MCP 的价值也得到了越来越多的人的认可，于是本篇文章将带你从 0 到 1 用 python 构建自己的 MCP  Server。

在开始之前，我们先了解一下 MCP 的主要传输协议。目前 MCP 支持两种主要的传输协议：

1. Stdio 传输协议：主要针对本地，需要在用户本地安装命令行工具，对运行环境有特定要求

2. SSE（Server-Sent Events）传输协议：主要针对云服务部署，基于 HTTP 长连接实现

我们将会分别开始构建基于 Stdio 传输协议的针对本地调用的 MCP Server，以及基于 SSE 传输协议的的部署在云服务器上的可远程调用的 MCP Server，以及对应的客户端。

目前市面上支持MCP的客户端主要有如Claude desktop，Cline，Cursor 等，由于claude封禁较严重，我们主要基于自建 Client，Cursor 和 Cline进行构建。

MCP Server:

1. Stdio 传输协议（本地）

2. SSE 传输协议 （远程）

MCP Client（客户端）：

1. 自建客户端（python)

2. Cursor

3. Cline

由于大模型受训练数据时间的限制，大模型本身固有的知识已满足不了互联网上最新实时最新的知识。特别是现在技术的快速迭代，langchain，llamaindex，autogen，agno，openai-agents-sdk，MCP 等众多等开源框架，它们都有自己的技术文档，而 MCP 最大的价值正是帮助大模型访问这些外部数据，所以我们将构建一个可以访问最新市场主流框架技术文档的 MCP Server。

01 第一步：环境配置

1.1 安装UV 包（相比于pip 更快）

在 MacOS/Linux 上：

curl -LsSf https://astral.sh/uv/install.sh | sh

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

更多安装方式详见：

https://docs.astral.sh/uv/getting-started/installation/

1.2 初始化项目

# Create a new directory for our projectuv init mcp-servercd mcp-server
# Create virtual environment and activate ituv venvsource .venv/bin/activate  # On Windows use: .venv\Scripts\activate
# Install dependenciesuv add "mcp[cli]" httpx

1.3 创建服务器实现文件

touch main.py

02 第二步：构建工具函数

为了让大模型能访问市面上主流框架的技术文档，我们主要通过用户输入的 query，结合指定 site 特定域名的谷歌搜索进行搜索相关网页，并对相关网页进行解析提取网页文本并返回。

其中谷歌搜索用的 serper.dev，我们需要到官网获取对应的 api key，新用户有 2500 次赠送查询次数。

2.1 构建相关文档映射字典（可根据自身开发需求，增加删减）：

docs_urls = {"langchain": "python.langchain.com/docs","llama-index": "docs.llamaindex.ai/en/stable","autogen":"microsoft.github.io/autogen/stable","agno":"docs.agno.com","openai-agents-sdk": "openai.github.io/openai-agents-python","mcp-doc":"modelcontextprotocol.io","camel-ai":"docs.camel-ai.org","crew-ai":"docs.crewai.com", ""}

2.2 构建 MCP 工具

async def search_web(query: str) -> dict | None:    payload = json.dumps({"q": query, "num": 3})
    headers = {        "X-API-KEY": os.getenv("SERPER_API_KEY"),        "Content-Type": "application/json",    }
    async with httpx.AsyncClient() as client:        try:            response = await client.post(                SERPER_URL, headers=headers, data=payload, timeout=30.0            )            response.raise_for_status()            return response.json()        except httpx.TimeoutException:            return {"organic": []}async def fetch_url(url: str):  async with httpx.AsyncClient() as client:        try:            response = await client.get(url, timeout=30.0)            soup = BeautifulSoup(response.text, "html.parser")            text = soup.get_text()            return text        except httpx.TimeoutException:            return "Timeout error"

@mcp.tool()  async def get_docs(query: str, library: str):  """    搜索给定查询和库的最新文档。    支持 langchain、llama-index、autogen、agno、openai-agents-sdk、mcp-doc、camel-ai 和 crew-ai。
    参数:    query: 要搜索的查询 (例如 "React Agent")    library: 要搜索的库 (例如 "agno")
    返回:    文档中的文本  """  if library not in docs_urls:    raise ValueError(f"Library {library} not supported by this tool")  query = f"site:{docs_urls[library]} {query}"  results = await search_web(query)  if len(results["organic"]) == 0:    return "No results found"  text = ""  for result in results["organic"]:    text += await fetch_url(result["link"])
  return text

03 第三步：封装 MCP Server (基于Stdio协议)

3.1 MCP Server (Studio)

# main.pyfrom mcp.server.fastmcp import FastMCPfrom dotenv import load_dotenvimport httpximport jsonimport osfrom bs4 import BeautifulSoup from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPfrom starlette.applications import Starlettefrom mcp.server.sse import SseServerTransportfrom starlette.requests import Requestfrom starlette.routing import Mount, Routefrom mcp.server import Serverimport uvicorn
load_dotenv()
mcp = FastMCP("Agentdocs")
USER_AGENT = "Agentdocs-app/1.0"SERPER_URL="https://google.serper.dev/search"
docs_urls = {"langchain": "python.langchain.com/docs","llama-index": "docs.llamaindex.ai/en/stable","autogen":"microsoft.github.io/autogen/stable","agno":"docs.agno.com","openai-agents-sdk": "openai.github.io/openai-agents-python","mcp-doc":"modelcontextprotocol.io","camel-ai":"docs.camel-ai.org","crew-ai":"docs.crewai.com"}
async def search_web(query: str) -> dict | None:    payload = json.dumps({"q": query, "num": 2})
    headers = {        "X-API-KEY": os.getenv("SERPER_API_KEY"),        "Content-Type": "application/json",    }
    async with httpx.AsyncClient() as client:        try:            response = await client.post(                SERPER_URL, headers=headers, data=payload, timeout=30.0            )            response.raise_for_status()            return response.json()        except httpx.TimeoutException:            return {"organic": []}async def fetch_url(url: str):  async with httpx.AsyncClient() as client:        try:            response = await client.get(url, timeout=30.0)            soup = BeautifulSoup(response.text, "html.parser")            text = soup.get_text()            return text        except httpx.TimeoutException:            return "Timeout error"
@mcp.tool()  async def get_docs(query: str, library: str):  """    搜索给定查询和库的最新文档。    支持 langchain、llama-index、autogen、agno、openai-agents-sdk、mcp-doc、camel-ai 和 crew-ai。
    参数:    query: 要搜索的查询 (例如 "React Agent")    library: 要搜索的库 (例如 "agno")
    返回:    文档中的文本  """  if library not in docs_urls:    raise ValueError(f"Library {library} not supported by this tool")  query = f"site:{docs_urls[library]} {query}"  results = await search_web(query)  if len(results["organic"]) == 0:    return "No results found"  text = ""  for result in results["organic"]:    text += await fetch_url(result["link"])
  return text

if __name__ == "__main__":    mcp.run(transport="stdio")

uv run main.py

3.2 客户端配置

3.2.1 基于cline

首先Visual studio Code 安装Cline 插件，然后进行配置MCP

{  "mcpServers": {      "mcp-server": {          "command": "uv",            "args": [              "--directory",              "<你的项目路径>",              "run",              "main.py"          ]      }  }}

成功绑定如图（左侧绿灯）：

3.2.2  基于Cursor

项目根目录创建 .cursor 文件夹，并创建 mcp.json 文件，如：

然后粘贴以下内容到 mcp.json

{  "mcpServers": {      "mcp-server": {          "command": "uv",            "args": [              "--directory",              "<你的项目路径>",              "run",              "main.py"          ]      }  }}

成功配置如图：

在Features开启MCP服务

通过对话它便通过MCP获取相关文档信息进行回答：

04 第四步：构建 SSE MCP Server (基于SSE协议)

4.1 封装 MCP Server

from mcp.server.fastmcp import FastMCPfrom dotenv import load_dotenvimport httpximport jsonimport osfrom bs4 import BeautifulSoup from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPfrom starlette.applications import Starlettefrom mcp.server.sse import SseServerTransportfrom starlette.requests import Requestfrom starlette.routing import Mount, Routefrom mcp.server import Serverimport uvicornload_dotenv()mcp = FastMCP("docs")USER_AGENT = "docs-app/1.0"SERPER_URL="https://google.serper.dev/search"docs_urls = {"langchain": "python.langchain.com/docs","llama-index": "docs.llamaindex.ai/en/stable","autogen":"microsoft.github.io/autogen/stable","agno":"docs.agno.com","openai-agents-sdk": "openai.github.io/openai-agents-python","mcp-doc":"modelcontextprotocol.io","camel-ai":"docs.camel-ai.org","crew-ai":"docs.crewai.com"}async def search_web(query: str) -> dict | None:    payload = json.dumps({"q": query, "num": 2})    headers = {        "X-API-KEY": os.getenv("SERPER_API_KEY"),        "Content-Type": "application/json",    }    async with httpx.AsyncClient() as client:        try:            response = await client.post(                SERPER_URL, headers=headers, data=payload, timeout=30.0            )            response.raise_for_status()            return response.json()        except httpx.TimeoutException:            return {"organic": []}async def fetch_url(url: str):  async with httpx.AsyncClient() as client:        try:            response = await client.get(url, timeout=30.0)            soup = BeautifulSoup(response.text, "html.parser")            text = soup.get_text()            return text        except httpx.TimeoutException:            return "Timeout error"@mcp.tool()  async def get_docs(query: str, library: str):  """    搜索给定查询和库的最新文档。    支持 langchain、llama-index、autogen、agno、openai-agents-sdk、mcp-doc、camel-ai 和 crew-ai。    参数:    query: 要搜索的查询 (例如 "React Agent")    library: 要搜索的库 (例如 "agno")    返回:    文档中的文本  """  if library not in docs_urls:    raise ValueError(f"Library {library} not supported by this tool")  query = f"site:{docs_urls[library]} {query}"  results = await search_web(query)  if len(results["organic"]) == 0:    return "No results found"  text = ""  for result in results["organic"]:    text += await fetch_url(result["link"])
  return text
## sse传输def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:    """Create a Starlette application that can server the provied mcp server with SSE."""    sse = SseServerTransport("/messages/")    async def handle_sse(request: Request) -> None:        async with sse.connect_sse(                request.scope,                request.receive,                request._send,  # noqa: SLF001        ) as (read_stream, write_stream):            await mcp_server.run(                read_stream,                write_stream,                mcp_server.create_initialization_options(),            )    return Starlette(        debug=debug,        routes=[            Route("/sse", endpoint=handle_sse),            Mount("/messages/", app=sse.handle_post_message),        ],    )if __name__ == "__main__":    mcp_server = mcp._mcp_server       import argparse    parser = argparse.ArgumentParser(description='Run MCP SSE-based server')    parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')    parser.add_argument('--port', type=int, default=8020, help='Port to listen on')    args = parser.parse_args()    # Bind SSE request handling to MCP server    starlette_app = create_starlette_app(mcp_server, debug=True)
    uvicorn.run(starlette_app, host=args.host, port=args.port)

uv run main.py --host 0.0.0.0 --port 8020

以上MCP server 代码直接在你的云服器跑即可。

4.2 构建 MCP Client

import asyncioimport jsonimport osfrom typing import Optionalfrom contextlib import AsyncExitStackimport timefrom mcp import ClientSessionfrom mcp.client.sse import sse_clientfrom openai import AsyncOpenAIfrom dotenv import load_dotenvload_dotenv()  # load environment variables from .envclass MCPClient:    def __init__(self):        # Initialize session and client objects        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"))    async def connect_to_sse_server(self, server_url: str):        """Connect to an MCP server running with SSE transport"""        # Store the context managers so they stay alive        self._streams_context = sse_client(url=server_url)        streams = await self._streams_context.__aenter__()        self._session_context = ClientSession(*streams)        self.session: ClientSession = await self._session_context.__aenter__()        # Initialize        await self.session.initialize()        # List available tools to verify connection        print("Initialized SSE client...")        print("Listing tools...")        response = await self.session.list_tools()        tools = response.tools        print("\nConnected to server with tools:", [tool.name for tool in tools])    async def cleanup(self):        """Properly clean up the session and streams"""        if self._session_context:            await self._session_context.__aexit__(None, None, None)        if self._streams_context:            await self._streams_context.__aexit__(None, None, None)    async def process_query(self, query: str) -> str:        """Process a query using OpenAI API and available tools"""        messages = [            {                "role": "user",                "content": query            }        ]        response = await self.session.list_tools()        available_tools = [{             "type": "function",            "function": {                "name": tool.name,                "description": tool.description,                "parameters": tool.inputSchema            }        } for tool in response.tools]        # Initial OpenAI API call        completion = await self.openai.chat.completions.create(            model=os.getenv("OPENAI_MODEL"),            max_tokens=1000,            messages=messages,            tools=available_tools        )        # Process response and handle tool calls        tool_results = []        final_text = []        assistant_message = completion.choices[0].message        if assistant_message.tool_calls:            for tool_call in assistant_message.tool_calls:                tool_name = tool_call.function.name                tool_args = json.loads(tool_call.function.arguments)                # Execute tool call                result = await self.session.call_tool(tool_name, tool_args)                tool_results.append({"call": tool_name, "result": result})                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
                # Continue conversation with tool results                messages.extend([                    {                        "role": "assistant",                        "content": None,                        "tool_calls": [tool_call]                    },                    {                        "role": "tool",                        "tool_call_id": tool_call.id,                        "content": result.content[0].text                    }                ])                print(f"Tool {tool_name} returned: {result.content[0].text}")                print("messages", messages)                # Get next response from OpenAI                completion = await self.openai.chat.completions.create(                    model=os.getenv("OPENAI_MODEL"),                    max_tokens=1000,                    messages=messages,                )                  if isinstance(completion.choices[0].message.content, (dict, list)):                    final_text.append(str(completion.choices[0].message.content))                else:                    final_text.append(completion.choices[0].message.content)        else:             if isinstance(assistant_message.content, (dict, list)):                final_text.append(str(assistant_message.content))            else:                final_text.append(assistant_message.content)        return "\n".join(final_text)    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 main():    if len(sys.argv) < 2:        print("Usage: uv run client.py <URL of SSE MCP server (i.e. http://localhost:8080/sse)>")        sys.exit(1)    client = MCPClient()    try:        await client.connect_to_sse_server(server_url=sys.argv[1])        await client.chat_loop()    finally:        await client.cleanup()if __name__ == "__main__":    import sys    asyncio.run(main())

uv run client.py http://0.0.0.0:8020/sse 

Client  日志：

Server 日志：

以上便是 python 从 0 到 1 搭建 MCP Server 以及 MCP Client 的完整教程。有不对的地方请多多指教，欢迎加入群聊进行交流。

完整代码：

https://github.com/GobinFan/python-mcp-server-client

相关参考资料：

1. https://www.youtube.com/watch?v=Ek8JHgZtmcI

2. https://serper.dev/

3. https://modelcontextprotocol.io/quickstart/server

4. https://modelcontextprotocol.io/quickstart/client

5.https://docs.cursor.com/context/model-context-protocol

关注公众号，用极客视角洞察未来！