🚀 MCP Python 实现

本项目是 MCP 的 Python 实现，提供了在 Python 环境下使用 MCP 协议的功能，可用于管理模型上下文，适用于 AI 和机器学习领域。

🚀 快速开始

安装

使用 pip 安装最新版本：

pip install python-mcp

💻 使用示例

基础用法

作为服务器

from mcp.server import Server, StdioServerParameters
from mcp.server.stdio import stdio_server
from mcp.types import NotificationOptions

async def run():
    server = Server()
    
    @server.list_prompts()
    async def handle_list_prompts() -> list[types.Prompt]:
        return [
            types.Prompt(
                name="example-prompt",
                description="示例提示模板",
                arguments=[
                    types.PromptArgument(
                        name="arg1", description="示例参数", required=True
                    )
                ],
            )
        ]

    @server.get_prompt()
    async def handle_get_prompt(
        name: str, arguments: dict[str, str] | None
    ) -> types.GetPromptResult:
        if name != "example-prompt":
            raise ValueError(f"未知提示：{name}")

        return types.GetPromptResult(
            description="示例提示",
            messages=[
                types.PromptMessage(
                    role="用户",
                    content=types.TextContent(type="text", text="示例提示文本"),
                )
            ],
        )

    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimentalCapabilities={},
                ),
            ),
        )


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

作为客户端

from mcp.client import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

# 创建连接到 MCP 服务器的参数
server_params = StdioServerParameters(
    command="python",  # 可执行文件路径
    args=["example_server.py"],  # 可选命令行参数
    env=None,  # 可选环境变量
)

# 可选：创建采样回调函数
async def handle_sampling_message(
    message: types.CreateMessageRequestParams,
) -> types.CreateMessageResult:
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(
            type="text",
            text="你好，世界！来自模型",
        ),
        model="gpt-3.5-turbo",
        stop_reason="endTurn",
    )

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write, sampling_callback=handle_sampling_message
        ) as session:
            # 获取提示列表
            await session.list_prompts()
            # 获取特定提示
            await session.get_prompt("example-prompt")

if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

📚 详细文档

MCP 基本操作

MCP 通知选项

MCP 协议

概念

MCP（Model Context Protocol）是一种用于管理模型上下文的协议，主要应用于 AI 和机器学习领域。

核心组件

1. 提示（Prompts）：定义了模型的行为和输入方式。
2. 资源（Resources）：提供给模型的数据和工具。
3. 工具（Tools）：扩展模型的功能。

实现步骤

1. 安装依赖：

   pip install python-mcp
   

2. 初始化服务器：

   from mcp.server import Server
   
   server = Server()
   

3. 注册提示处理函数：

   @server.list_prompts()
   async def handle_list_prompts() -> list[types.Prompt]:
       return [...]  # 返回提示列表
   

4. 启动服务器：

   async with stdio_server() as (read_stream, write_stream):
       await server.run(...)
   

📖 参考资料

• MCP 官方文档
• Python MCP 仓库

以上是 MCP Python 实现的完整指南，涵盖了安装、示例代码、表格说明和协议概述。