MCP代码示例

三月 15, 2025

MCP架构分为Server端和Client端，Server端负责对第三方工具调用或者业务逻辑代码实现，Client端主要是提供SDK给调用放使用用来执行Server端提供的工具指令。

有一点需要注意，MCP提供的list_tools暴漏的出来的方法定义的结构，是function_calling规范并不一致，如果是要与function_calling结合使用，注意进行结构转换。

MCP架构.drawio

MCP Server

初始化环境

# 初始化项目
uv init mcp-server-demo
cd mcp-server-demo

# 创建虚拟环境
uv venv
source .venv/bin/activate

# 添加依赖
uv add "mcp[cli]"
# uv sync

编写Server端示例代码（server.py）

使用FastMCP定义一个最简单的函数，并通过@mcp.tool() 装饰器进行服务暴漏

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("mcp-server-demo")


# 通过装饰器定义工具函数
@mcp.tool()
async def get_user_info(name: str) -> str:
    """获取用户介绍信息

    Args:
        name: 用户姓名
    """
    return f"我叫{name},今年20岁，目前就读于家里蹲大学计算机科学与技术专业，大三学生。自小对科技充满好奇，尤其对编程和算法设计有着浓厚的兴趣。在校期间，我积极参与各类学术活动和项目实践，曾参与开发校园管理系统，并在团队中担任核心开发角色，积累了丰富的项目经验。此外，我还担任了学校编程社团的副社长，组织并参与了多次编程竞赛和技术分享会，提升了我的组织能力和沟通技巧。课余时间，我喜欢阅读科技类书籍和文章，关注前沿技术动态，尤其是人工智能和大数据领域。未来，我计划继续深造，攻读硕士学位，致力于成为一名优秀的软件工程师，为科技发展贡献自己的力量。我相信，通过不断学习和实践，我能够在未来的职业生涯中取得更大的成就。"


def main():
    """运行入口"""
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

使用python直接运行，通过stdio输入指令进行测试

或者通过inspector进行测试

npx @modelcontextprotocol/inspector python server.py

将MCP Server安装到本地

先修改pyproject.toml 增加入口配置project.scripts

[project]
name = "mcp-server-demo"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "mcp[cli]>=1.4.1",
]
[project.scripts]
mcp-server-demo = "server:main"

通过uv指令安装到本地

# 安装到本地
uv tool -n install .
# 查看是否安装成功
uv tool list
# 通过指令运行(等同于 uvx mcp-server-demo)
uv tool run mcp-server-demo

通过指令或者inspector进行测试

MCP Client

初始化环境

# 初始化项目
uv init mcp-client-demo
cd mcp-client-demo

# 创建虚拟环境
uv venv
source .venv/bin/activate

# 添加依赖
uv add mcp
# uv sync

编写Client端代码（client.py）

import asyncio
from contextlib import AsyncExitStack
from typing import Optional
from mcp import ClientSession, StdioServerParameters, stdio_client


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect(self, command: str, args: list[str]):
        """创建对Mcp Server的连接"""
        # 创建stdio输入输出流
        server_params = StdioServerParameters(command=command, args=args)
        read_stream, write_stream = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )

        # 创建基于stdio的session
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        await self.session.initialize()
        # 返回session，用于操作MCP Server
        return self.session

    async def close(self):
        """关闭连接"""
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        # 创建连接
        session = await client.connect("uvx", ["mcp-server-demo"])
        # 列出所有可用工具
        print("===================list_tools=====================")
        result = await session.list_tools()
        for t in result.tools:
            print(t.model_dump_json(indent=2))

        # 调用执行工具函数的示例
        method = "get_user_info"
        args = {"name": "李四"}
        print(f"===================call {method}({args})=====================")
        result = await session.call_tool(method, args)
        for c in result.content:
            print(c.text)

    finally:
        await client.close()


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

‍