MCP 进阶专题

from mcp.server.fastmcp import FastMCP, Context
from mcp.types import SamplingMessage, TextContent

mcp = FastMCP('sampling-demo')

@mcp.tool()
async def summarize_text(text: str, ctx: Context) -> str:
    '''使用 LLM 对给定文本生成摘要（通过 Sampling 机制）'''
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role='user',
                content=TextContent(
                    type='text',
                    text=f'请用两句话概括以下内容：\n\n{text}'
                )
            )
        ],
        max_tokens=200
    )
    return result.content.text

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

from mcp.server.fastmcp import FastMCP, Context
import asyncio

mcp = FastMCP('notification-demo')

@mcp.tool()
async def process_files(file_count: int, ctx: Context) -> str:
    '''模拟批量处理文件，展示日志和进度通知'''
    await ctx.session.send_log_message(level='info', data=f'开始处理 {file_count} 个文件')

    for i in range(file_count):
        await asyncio.sleep(0.5)
        await ctx.session.send_log_message(level='debug', data=f'正在处理文件 {i + 1}/{file_count}')

        if ctx.meta and ctx.meta.progress_token:
            await ctx.session.send_progress(
                progress_token=ctx.meta.progress_token,
                progress=i + 1,
                total=file_count
            )

    await ctx.session.send_log_message(level='info', data=f'所有 {file_count} 个文件处理完成')
    return f'成功处理 {file_count} 个文件'

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

from mcp.server.fastmcp import FastMCP, Context
from pathlib import Path

mcp = FastMCP('roots-demo')

def is_path_within_roots(file_path: str, roots: list) -> bool:
    '''检查文件路径是否在允许的 Roots 范围内'''
    target = Path(file_path).resolve()
    for root in roots:
        root_path = Path(root.uri.replace('file://', '')).resolve()
        try:
            target.relative_to(root_path)
            return True
        except ValueError:
            continue
    return False

@mcp.tool()
async def safe_read_file(file_path: str, ctx: Context) -> str:
    '''安全地读取文件，只允许访问客户端声明的 Roots 范围内的路径'''
    roots_result = await ctx.session.list_roots()
    roots = roots_result.roots

    if not roots:
        return '错误：客户端未声明任何 Root，无法访问文件'

    if not is_path_within_roots(file_path, roots):
        return f'错误：路径 {file_path} 不在允许的范围内。'

    with open(file_path, 'r', encoding='utf-8') as f:
        return f.read()

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

// Request — 客户端调用工具
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tools/call",
  "params": { "name": "add_numbers", "arguments": { "a": 3, "b": 5 } }
}

// Response — 服务器成功响应
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": { "content": [{ "type": "text", "text": "8" }] }
}

// Notification — 服务器推送日志（无 id）
{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": { "level": "info", "data": "处理完成" }
}

// Error Response — 请求处理失败
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "error": { "code": -32602, "message": "参数类型错误" }
}

from mcp.server.fastmcp import FastMCP, Context
import asyncio

mcp = FastMCP('http-server')

@mcp.tool()
async def long_task(steps: int, ctx: Context) -> str:
    '''模拟长时间运行的任务，通过 SSE 推送进度'''
    for i in range(steps):
        await asyncio.sleep(1)
        await ctx.session.send_log_message(level='info', data=f'步骤 {i + 1}/{steps} 完成')
    return f'任务完成，共执行 {steps} 步'

if __name__ == '__main__':
    mcp.run(
        transport='streamable-http',
        host='0.0.0.0',
        port=8000
    )

# 启动后访问：http://localhost:8000/mcp

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    server_url = 'http://localhost:8000/mcp'

    async with streamablehttp_client(server_url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            session.on_log_message = lambda msg: print(f'[服务器日志] {msg.data}')
            result = await session.call_tool('long_task', {'steps': 3})
            print(f'结果: {result.content[0].text}')

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