Model Context Protocol 入门

# 方式一：使用 pip
pip install mcp

# 方式二：使用 uv（推荐）
uv init my-mcp-server
cd my-mcp-server
uv add mcp

# server.py 最小起步代码
from mcp.server.fastmcp import FastMCP

mcp = FastMCP('my-server')

# 稍后在这里添加 tools、resources、prompts

if __name__ == '__main__':
    mcp.run()

# 开发模式启动（自动打开 Inspector）
mcp dev server.py

# 或使用 uv run
uv run mcp dev server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP('my-server')

@mcp.tool()
def add_numbers(a: int, b: int) -> int:
    '''将两个整数相加并返回结果。'''
    return a + b

@mcp.tool()
def read_file(path: str) -> str:
    '''读取指定路径的文件内容并以字符串形式返回。
    
    Args:
        path: 要读取的文件的绝对路径
    '''
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

if __name__ == '__main__':
    mcp.run()

# 启动服务器并打开 Inspector
mcp dev server.py

# 终端输出示例：
# MCP Inspector is up and running at http://localhost:5173
# Starting MCP server my-server...

// 请求（Inspector 发送给服务器）
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "add_numbers",
    "arguments": { "a": 3, "b": 5 }
  }
}

// 响应（服务器返回）
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{ "type": "text", "text": "8" }]
  }
}

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

async def main():
    # 描述如何启动服务器
    server_params = StdioServerParameters(
        command='python',
        args=['server.py']
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            # 列出所有工具
            tools = await session.list_tools()
            print('可用工具：')
            for tool in tools.tools:
                print(f'  - {tool.name}: {tool.description}')

            # 调用工具
            result = await session.call_tool('add_numbers', {'a': 10, 'b': 32})
            print(f'\n调用结果：{result.content[0].text}')

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

python client.py

# 预期输出：
# 可用工具：
#   - add_numbers: 将两个整数相加并返回结果。
#   - read_file: 读取指定路径的文件内容...
#
# 调用结果：42

from mcp.server.fastmcp import FastMCP

mcp = FastMCP('my-server')

# 静态资源：固定 URI
@mcp.resource('resource://config/app')
def get_app_config() -> str:
    '''返回应用配置文件内容'''
    with open('config.json', 'r', encoding='utf-8') as f:
        return f.read()

# 动态资源：URI 模板（使用 {参数} 占位符）
@mcp.resource('resource://files/{filename}')
def get_file(filename: str) -> str:
    '''读取 data 目录下指定文件的内容'''
    safe_path = f'./data/{filename}'
    with open(safe_path, 'r', encoding='utf-8') as f:
        return f.read()

if __name__ == '__main__':
    mcp.run()

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

async def main():
    server_params = StdioServerParameters(command='python', args=['server.py'])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出所有资源
            resources = await session.list_resources()
            print('可用资源：')
            for r in resources.resources:
                print(f'  URI: {r.uri}')
                print(f'  描述: {r.description}')

            # 读取具体资源
            content = await session.read_resource('resource://config/app')
            print(f'\n配置内容：')
            for item in content.contents:
                print(item.text)

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

from mcp.server.fastmcp import FastMCP
from mcp.types import PromptMessage, TextContent

mcp = FastMCP('my-server')

@mcp.prompt()
def code_review(code: str, language: str = 'python') -> list[PromptMessage]:
    '''生成代码审查提示模板'''
    return [
        PromptMessage(
            role='user',
            content=TextContent(
                type='text',
                text=f'请审查以下 {language} 代码，指出潜在问题并给出改进建议：\n\n```{language}\n{code}\n```'
            )
        )
    ]

@mcp.prompt()
def summarize_document(document: str, style: str = '简洁') -> list[PromptMessage]:
    '''生成文档摘要提示模板，支持指定摘要风格'''
    return [
        PromptMessage(
            role='user',
            content=TextContent(
                type='text',
                text=f'请以{style}的风格为以下文档写一段摘要：\n\n{document}'
            )
        )
    ]

if __name__ == '__main__':
    mcp.run()

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

async def main():
    server_params = StdioServerParameters(command='python', args=['server.py'])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出所有 Prompts
            prompts = await session.list_prompts()
            print('可用 Prompts：')
            for p in prompts.prompts:
                print(f'  名称: {p.name}')
                print(f'  描述: {p.description}')
                print(f'  参数: {[arg.name for arg in (p.arguments or [])]}')

            # 使用 code_review prompt
            result = await session.get_prompt(
                'code_review',
                arguments={
                    'code': 'def add(a, b):\n    return a + b',
                    'language': 'python'
                }
            )

            print('\n生成的消息模板：')
            for msg in result.messages:
                print(f'[{msg.role}] {msg.content.text[:100]}...')

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