将 FastMCP 与 Claude Code 集成

Claude Code 是 Anthropic 官方支持 MCP 服务器的命令行工具 (CLI)，它允许用户通过自定义 FastMCP 服务器扩展 Claude 的功能。这种集成实现了强大的开发工作流，让 Claude 能够访问你的工具、资源和特定领域的功能。

为什么要与 Claude Code 集成？

Claude Code 的集成提供了以下几个引人注目的优势：

增强开发工作流：Claude 可以直接访问你的自定义工具。
上下文感知协助：Claude 能够读取你的项目资源和配置。
自动化任务：通过你的 MCP 工具执行复杂操作。
本地和远程支持：可连接本地开发服务器和已部署的服务。
实时交互：动态工具发现和执行。

设置 FastMCP 服务器

创建开发服务器

让我们创建一个专为开发工作流设计的综合性 FastMCP 服务器：

# dev_server.py
import os
import subprocess
import json
from pathlib import Path
from datetime import datetime
from fastmcp import FastMCP, Context

mcp = FastMCP(
    name="Development Assistant",
    instructions="""
    此服务器提供开发工具，包括：
    - 项目分析和文件操作
    - Git 操作和仓库管理
    - 代码质量检查和测试
    - 环境和依赖管理
    """
)

@mcp.tool
def run_command(command: str, cwd: str = ".") -> dict:
    """执行一个 shell 命令并返回结果。"""
    try:
        result = subprocess.run(
            command.split(),
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "command": command,
            "returncode": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {"error": "命令执行超时（30 秒）"}
    except Exception as e:
        return {"error": str(e)}

@mcp.tool
def analyze_project_structure(path: str = ".") -> dict:
    """分析项目目录结构。"""
    project_path = Path(path)

    if not project_path.exists():
        return {"error": f"路径 {path} 不存在"}

    structure = {
        "root": str(project_path.absolute()),
        "files": [],
        "directories": [],
        "python_files": [],
        "config_files": [],
        "total_size": 0
    }

    config_extensions = {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg"}

    for item in project_path.rglob("*"):
        if item.is_file():
            size = item.stat().st_size
            structure["total_size"] += size

            file_info = {
                "path": str(item.relative_to(project_path)),
                "size": size,
                "modified": datetime.fromtimestamp(item.stat().st_mtime).isoformat()
            }

            structure["files"].append(file_info)

            if item.suffix == ".py":
                structure["python_files"].append(file_info)
            elif item.suffix in config_extensions:
                structure["config_files"].append(file_info)

        elif item.is_dir():
            structure["directories"].append(str(item.relative_to(project_path)))

    return structure

@mcp.tool
async def check_code_quality(file_path: str, ctx: Context) -> dict:
    """使用各种 linting 工具检查代码质量。"""
    await ctx.info(f"正在检查 {file_path} 的代码质量...")

    if not os.path.exists(file_path):
        return {"error": f"文件 {file_path} 未找到"}

    results = {"file": file_path, "checks": {}}

    # 使用不同的工具检查
    tools = {
        "flake8": ["flake8", file_path],
        "pylint": ["pylint", file_path, "--output-format=json"],
        "mypy": ["mypy", file_path]
    }

    for tool, command in tools.items():
        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=15
            )
            results["checks"][tool] = {
                "returncode": result.returncode,
                "output": result.stdout,
                "errors": result.stderr
            }
        except (subprocess.TimeoutExpired, FileNotFoundError):
            results["checks"][tool] = {"error": f"{tool} 不可用或超时"}

    return results

@mcp.resource("project://config")
def get_project_config() -> dict:
    """从各种配置文件中获取项目配置。"""
    config = {"found_files": [], "configurations": {}}

    config_files = [
        "pyproject.toml", "setup.py", "requirements.txt", 
        "package.json", "Dockerfile", ".env", "config.yaml"
    ]

    for config_file in config_files:
        if os.path.exists(config_file):
            config["found_files"].append(config_file)
            try:
                with open(config_file, 'r') as f:
                    content = f.read()
                config["configurations"][config_file] = {
                    "content": content[:1000] + ("..." if len(content) > 1000 else ""),
                    "size": len(content)
                }
            except Exception as e:
                config["configurations"][config_file] = {"error": str(e)}

    return config

@mcp.resource("git://status")
def get_git_status() -> dict:
    """获取当前 Git 仓库状态。"""
    try:
        # 检查是否在 git 仓库中
        subprocess.run(["git", "rev-parse", "--git-dir"], 
                      check=True, capture_output=True)

        # 获取各种 git 信息
        commands = {
            "branch": ["git", "branch", "--show-current"],
            "status": ["git", "status", "--porcelain"],
            "last_commit": ["git", "log", "-1", "--pretty=format:%H|%an|%ad|%s"],
            "remote": ["git", "remote", "-v"]
        }

        result = {}
        for key, command in commands.items():
            try:
                output = subprocess.run(
                    command, capture_output=True, text=True, check=True
                ).stdout.strip()
                result[key] = output
            except subprocess.CalledProcessError:
                result[key] = "不可用"

        return result

    except subprocess.CalledProcessError:
        return {"error": "当前目录不是一个 Git 仓库"}

if __name__ == "__main__":
    mcp.run(transport="http", host="127.0.0.1", port=8000)

连接到 Claude Code

运行服务器

首先，启动 FastMCP 服务器：

# 以 HTTP 模式启动服务器
python dev_server.py

你的服务器将在 http://127.0.0.1:8000/mcp/ 运行。

将服务器添加到 Claude Code

使用 Claude Code 的 MCP 管理命令添加你的服务器：

# 添加你的 FastMCP 服务器
claude mcp add dev-assistant --transport http http://localhost:8000/mcp/

# 验证服务器是否已连接
claude mcp list

备选方案：STDIO 传输

对于本地开发，你可能更喜欢 STDIO 传输：

# dev_server_stdio.py
# ... (与上面的服务器代码相同)

if __name__ == "__main__":
    mcp.run()  # 默认使用 STDIO

将其添加到 Claude Code：

claude mcp add dev-assistant --transport stdio "python dev_server_stdio.py"

使用 Claude Code 访问你的服务器

连接后，你可以通过 Claude Code 与你的 MCP 服务器交互：

基本工具使用

# Claude 会自动发现并使用你的工具
claude "分析我当前的项目结构"

# Claude 将调用你的 analyze_project_structure 工具
# 响应将包含文件数量、大小和组织结构

资源访问

Claude 可以通过 @ 提及来访问服务器的资源：

# 访问项目配置
claude "这个项目有什么配置文件？使用 @dev-assistant:project://config"

# 检查 git 状态
claude "当前的 git 状态是什么？使用 @dev-assistant:git://status"

复杂工作流

结合多个工具进行复杂操作：

claude "请检查 src/ 目录下所有 Python 文件的代码质量并总结问题"

# Claude 将：
# 1. 使用 analyze_project_structure 找到 Python 文件
# 2. 对每个文件使用 check_code_quality
# 3. 总结结果

高级配置示例

带有数据库工具的开发服务器

# db_server.py
import sqlite3
from fastmcp import FastMCP

mcp = FastMCP("Database Tools")

@mcp.tool
def query_database(query: str, db_path: str = "app.db") -> dict:
    """对数据库执行 SQL 查询。"""
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        cursor.execute(query)

        if query.strip().upper().startswith("SELECT"):
            results = cursor.fetchall()
            columns = [description[0] for description in cursor.description]
            return {
                "columns": columns,
                "rows": results,
                "count": len(results)
            }
        else:
            conn.commit()
            return {"affected_rows": cursor.rowcount}

    except Exception as e:
        return {"error": str(e)}
    finally:
        conn.close()

@mcp.resource("db://schema")
def get_database_schema() -> dict:
    """获取数据库模式信息。"""
    # 获取数据库模式的实现
    pass

if __name__ == "__main_":
    mcp.run(transport="http", port=8001)

API 测试服务器

# api_server.py
import requests
from fastmcp import FastMCP, Context

mcp = FastMCP("API Testing Tools")

@mcp.tool
async def test_api_endpoint(url: str, method: str = "GET", 
                          headers: dict = None, data: dict = None,
                          ctx: Context = None) -> dict:
    """测试 API 端点并返回详细结果。"""
    await ctx.info(f"正在测试 {method} {url}...")

    try:
        response = requests.request(
            method=method.upper(),
            url=url,
            headers=headers or {},
            json=data,
            timeout=10
        )

        return {
            "url": url,
            "method": method.upper(),
            "status_code": response.status_code,
            "headers": dict(response.headers),
            "response_time": response.elapsed.total_seconds(),
            "body": response.text[:1000] if response.text else None,
            "success": 200 <= response.status_code < 300
        }

    except Exception as e:
        return {
            "url": url,
            "method": method.upper(),
            "error": str(e),
            "success": False
        }

if __name__ == "__main__":
    mcp.run(transport="http", port=8002)

多服务器设置

你可以将多个 FastMCP 服务器连接到 Claude Code：

# 添加多个服务器
claude mcp add dev-tools --transport http http://localhost:8000/mcp/
claude mcp add db-tools --transport http http://localhost:8001/mcp/
claude mcp add api-tools --transport http http://localhost:8002/mcp/

# Claude 现在可以访问所有服务器
claude "检查项目结构，然后测试配置中定义的 API 端点"

常见问题故障排除

服务器无响应

# 检查服务器是否正在运行
curl http://localhost:8000/mcp/

# 移除并重新添加服务器
claude mcp remove dev-assistant
claude mcp add dev-assistant --transport http http://localhost:8000/mcp/

未找到工具

确保你的服务器已正确注册并且工具已用装饰器标记：

# 确保工具已正确装饰
@mcp.tool  # 此装饰器是必需的
def my_tool():
    pass

资源访问问题

检查资源 URI 格式：

# 正确的格式
claude "使用 @server-name:resource://uri"

# 错误的格式: @server-name/resource://uri

Claude Code 集成最佳实践

清晰的工具名称：使用描述性强、不言自明的工具名称。
良好的文档：包含全面的文档字符串。
错误处理：提供有意义的错误消息。
资源组织：使用逻辑 URI 方案来组织资源。
性能：保持工具执行时间合理。
安全性：验证输入并限制危险操作。

后续步骤

将 FastMCP 服务器集成到 Claude Code 后，你可以：

构建特定领域的开发助手。
自动化复杂的项目管理任务。
创建自定义代码分析和质量工具。
开发专门的测试和部署工作流。

在最后一部分，我们将探讨 FastMCP 服务器的高级模式、生产部署和扩展策略。

FastMCP：构建MCP服务器与客户端的完整指南

将 FastMCP 与 Claude 代码集成