1.3 第一个MCP服务

在本章中，我们将创建一个简单但功能完整的MCP服务。通过构建这个服务，您将了解如何定义资源、工具和提示，以及如何使用不同的传输方式与服务器交互。

创建简单的"Hello World"服务器

让我们从最基本的MCP服务器开始。创建一个名为hello_world.py的文件，并添加以下代码：

from mcp.server.fastmcp import FastMCP

# 创建一个MCP服务器
mcp = FastMCP("Hello World")

# 添加一个简单的工具
@mcp.tool()
def say_hello(name: str) -> str:
    """向用户问好"""
    return f"Hello, {name}! Welcome to MCP."

# 运行服务器
if __name__ == "__main__":
    mcp.run()

这个简单的服务器有以下几个部分：

导入FastMCP类 - 这是创建MCP服务器的主要接口
创建一个命名为"Hello World"的服务器实例
定义一个名为say_hello的工具，它接受一个字符串参数并返回一个问候语
启动服务器

您可以通过运行以下命令启动这个服务器：

python hello_world.py

默认情况下，服务器使用stdio传输方式运行，这意味着它通过标准输入/输出与客户端通信。

定义基本资源和工具

接下来，让我们创建一个更复杂的服务器，包含资源、工具和提示。创建一个名为note_taking.py的文件：

from datetime import datetime
from typing import Dict, List

from mcp.server.fastmcp import Context, FastMCP

# 创建一个MCP服务器
mcp = FastMCP("Note Taking App")

# 模拟数据存储
notes_db: Dict[str, List[str]] = {}

# 定义资源

@mcp.resource("notes://{user_id}")
def get_user_notes(user_id: str) -> str:
    """获取用户的所有笔记"""
    if user_id not in notes_db or not notes_db[user_id]:
        return "该用户没有笔记。"
    
    notes = notes_db[user_id]
    return "\n\n".join([f"Note {i+1}: {note}" for i, note in enumerate(notes)])

@mcp.resource("help://notes")
def get_help() -> str:
    """获取笔记应用的帮助信息"""
    return """
    笔记应用使用指南:
    
    1. 使用 'add_note' 工具添加新笔记
    2. 使用 'notes://{user_id}' 资源查看笔记
    3. 使用 'delete_note' 工具删除笔记
    """

# 定义工具

@mcp.tool()
def add_note(user_id: str, content: str) -> str:
    """添加一条新笔记
    
    参数:
        user_id: 用户ID
        content: 笔记内容
    
    返回:
        添加结果消息
    """
    if user_id not in notes_db:
        notes_db[user_id] = []
    
    # 添加时间戳
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    full_content = f"[{timestamp}] {content}"
    
    notes_db[user_id].append(full_content)
    return f"笔记已添加，当前有 {len(notes_db[user_id])} 条笔记。"

@mcp.tool()
def delete_note(user_id: str, note_index: int) -> str:
    """删除指定的笔记
    
    参数:
        user_id: 用户ID
        note_index: 笔记索引（从1开始）
    
    返回:
        删除结果消息
    """
    if user_id not in notes_db or not notes_db[user_id]:
        return "找不到笔记可删除。"
    
    if note_index < 1 or note_index > len(notes_db[user_id]):
        return f"无效的笔记索引。有效范围: 1-{len(notes_db[user_id])}"
    
    deleted_note = notes_db[user_id].pop(note_index - 1)
    return f"已删除笔记: {deleted_note}"

@mcp.tool()
def search_notes(user_id: str, keyword: str) -> str:
    """搜索笔记
    
    参数:
        user_id: 用户ID
        keyword: 要搜索的关键词
    
    返回:
        匹配的笔记列表
    """
    if user_id not in notes_db or not notes_db[user_id]:
        return "该用户没有笔记可搜索。"
    
    matches = [note for note in notes_db[user_id] if keyword.lower() in note.lower()]
    
    if not matches:
        return f"没有找到包含 '{keyword}' 的笔记。"
    
    return "\n\n".join([f"Match {i+1}: {note}" for i, note in enumerate(matches)])

# 定义提示

@mcp.prompt("create-note")
def create_note_prompt(user_id: str) -> str:
    """创建新笔记的提示"""
    return f"""
    为用户 '{user_id}' 创建一个新笔记。
    
    请描述您想记录的内容，我会为您保存。
    """

@mcp.prompt("view-notes")
def view_notes_prompt(user_id: str) -> str:
    """查看笔记的提示"""
    return f"""
    下面是用户 '{user_id}' 的所有笔记:
    
    {{{{ notes://{user_id} }}}}
    
    您可以添加新笔记或删除现有笔记。
    """

# 运行服务器
if __name__ == "__main__":
    mcp.run()

这个笔记应用服务器包含以下功能：

资源:
- notes://{user_id} - 获取指定用户的所有笔记
- help://notes - 获取应用的帮助信息
工具:
- add_note - 添加新笔记
- delete_note - 删除笔记
- search_notes - 搜索笔记
提示:
- create-note - 创建笔记的提示模板
- view-notes - 查看笔记的提示模板

同样，您可以运行这个服务器：

python note_taking.py

使用不同的传输方式

MCP支持多种传输方式，默认使用stdio。让我们修改服务器以支持SSE (Server-Sent Events) 传输：

if __name__ == "__main__":
    # 使用SSE传输，在端口8000上监听
    mcp.run(transport="sse")

使用SSE传输时，服务器将在HTTP端口上运行一个Web服务，允许通过网络连接到它。

您也可以在启动服务器时指定传输方式：

# 使用stdio（默认）
python note_taking.py

# 使用SSE
python note_taking.py --transport sse

与服务器进行基本交互

现在，让我们创建一个客户端脚本来与笔记应用交互。创建一个名为note_client.py的文件：

Stdio 客户端

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

async def run_stdio_client():
    server_params = StdioServerParameters(
        command="python",
        args=["note_taking.py"],
    )
    
    print("连接到笔记应用服务器...")
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            print("连接成功！")
            
            # 获取用户ID
            user_id = "user123"
            
            # 获取帮助信息
            help_content, _ = await session.read_resource("help://notes")
            print(f"\n帮助信息:\n{help_content}")
            
            # 添加笔记
            result = await session.call_tool(
                "add_note", 
                arguments={"user_id": user_id, "content": "记得购买牛奶和鸡蛋"}
            )
            print(f"\n添加笔记结果: {result.content[0].text}")
            
            result = await session.call_tool(
                "add_note", 
                arguments={"user_id": user_id, "content": "约会时间：周五晚上7点"}
            )
            print(f"添加第二条笔记结果: {result.content[0].text}")
            
            # 查看所有笔记
            notes, _ = await session.read_resource(f"notes://{user_id}")
            print(f"\n当前笔记:\n{notes}")
            
            # 搜索笔记
            result = await session.call_tool(
                "search_notes", 
                arguments={"user_id": user_id, "keyword": "约会"}
            )
            print(f"\n搜索结果: {result.content[0].text}")
            
            # 删除笔记
            result = await session.call_tool(
                "delete_note", 
                arguments={"user_id": user_id, "note_index": 1}
            )
            print(f"\n删除笔记结果: {result.content[0].text}")
            
            # 再次查看所有笔记
            notes, _ = await session.read_resource(f"notes://{user_id}")
            print(f"\n删除后的笔记:\n{notes}")
            
            # 获取提示
            prompt = await session.get_prompt(
                "view-notes", arguments={"user_id": user_id}
            )
            print(f"\n查看笔记提示:\n{prompt.messages[0].content.text}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_stdio_client())

SSE 客户端

如果您使用SSE传输，客户端代码稍有不同：

from mcp import ClientSession
from mcp.client.sse import sse_client

async def run_sse_client():
    server_url = "http://localhost:8000/sse"
    
    print(f"连接到笔记应用服务器 {server_url}...")
    async with sse_client(server_url) as (read, write):
        async with ClientSession(read, write) as session:
            # 与stdio客户端的其余代码相同
            await session.initialize()
            print("连接成功！")
            
            # ... 其余代码与stdio客户端相同 ...

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_sse_client())

故障排除和常见错误

在使用MCP时，您可能会遇到一些常见问题。以下是一些故障排除技巧：

1. 连接问题

如果客户端无法连接到服务器：

确保服务器正在运行
检查传输方式是否匹配（stdio vs. SSE）
对于SSE，检查URL和端口是否正确
检查防火墙设置

2. 资源不可用

如果资源读取失败：

确保资源名称和参数正确
检查资源是否在服务器中定义
注意URI参数格式，特别是对类型敏感的参数

3. 工具调用失败

如果工具调用返回错误：

检查工具名称是否正确
确认所有必需参数都已提供
验证参数类型是否正确（例如，数字vs字符串）

4. 提示问题

如果提示获取失败：

确认提示名称是否正确
检查提示参数是否完整
注意提示内容中的资源引用格式

5. 调试技巧

添加更多的日志输出
使用MCP开发工具检查服务器：mcp dev note_taking.py
检查服务器配置和传输选项
验证依赖项版本与MCP兼容

小结

在本章中，我们创建了第一个功能完整的MCP服务 - 一个简单的笔记应用。我们学习了如何：

创建MCP服务器实例
定义和实现资源
创建工具函数
设计提示模板
使用不同的传输方式（stdio和SSE）
构建客户端应用程序与服务器交互
处理常见问题

通过这个示例，您已经了解了MCP的基本工作原理和用法。在下一章中，我们将探索客户端应用程序的更多细节，包括如何处理连接、会话和错误。