Next MCP (Keshav Varad)

下一个MCP

 ](https://badge.fury.io/py/nextmcp) ](https://pypi.org/project/nextmcp/)  

具有最小样板的生产级MCP服务器工具包

NextMCP是一个基于FastMCP构建的Python SDK，为构建MCP（模型上下文协议）服务器提供了对开发人员友好的体验。受Next.js的启发，它提供了最少的设置、强大的中间件和丰富的CLI，用于快速开发。

特性

• 完整的MCP规范 -完全支持工具、提示和资源原语
• 基于惯例的结构 -Next.js启发的基于文件的组织，具有自动发现功能
• 零配置设置 -单线 NextMCP.from_config() 用于即时项目设置
• 自动发现 -从目录结构中自动发现和注册基元
• 生产部署 -一个命令部署到Docker、Railway、Render、Fly.io并进行健康检查
• 健康检查 -为Kubernetes和云平台内置的活性和就绪性探测器
• 优雅地关闭 -通过SIGTERM/SIGINT处理和资源清理来清理终止
• 最小沸腾板 -只需几行代码即可开始
• 基于装饰器的API -使用简单的装饰器注册工具、提示和资源
• 异步支持 -跨所有原语完全支持async/await
• 参数完成 -对提示参数和资源模板的智能建议
• 资源订阅 -资源更改时的实时通知
• WebSocket传输 -交互式应用程序的实时双向通信
• 身份验证和授权 -内置支持API密钥、JWT、会话、OAuth 2.0和RBAC
• OAuth 2.0支持 -使用GitHub和Google提供商、PKCE、会话管理完成OAuth实现
• 全局和原始特定中间件 -添加日志记录、身份验证、速率限制、缓存等
• 丰富的CLI -使用以下工具构建项目、运行服务器、部署和生成文档 mcp 命令
• 配置管理 -支持 .env、YAML配置文件和环境变量
• 架构验证 -可选Pydantic集成，用于类型安全输入
• 度量与监控 -支持自定义指标的Prometheus兼容指标
• 生产就绪 -内置错误处理、日志记录和全面测试

安装

基本安装

pip install nextmcp

具有可选依赖关系

# CLI tools (recommended)
pip install nextmcp[cli]

# Configuration support
pip install nextmcp[config]

# Schema validation with Pydantic
pip install nextmcp[schema]

# WebSocket transport
pip install nextmcp[websocket]

# OAuth authentication (included in base by default via aiohttp)
pip install nextmcp[oauth]

# Everything
pip install nextmcp[all]

# Development dependencies
pip install nextmcp[dev]

快速开始

NextMCP提供了两种方法： 基于公约 （推荐）用于可扩展的项目，以及 手册 对于简单的用例。

基于公约的方法（推荐）

非常适合具有多种工具、提示和资源的项目。使用基于文件的组织进行自动发现。

1.创建项目结构

my-blog-server/
├── nextmcp.config.yaml
├── server.py
├── tools/
│   ├── __init__.py
│   └── posts.py
├── prompts/
│   ├── __init__.py
│   └── workflows.py
└── resources/
    ├── __init__.py
    └── blog_resources.py

2.配置您的项目

# nextmcp.config.yaml
name: blog-server
version: 1.0.0
description: A blog management MCP server

auto_discover: true

discovery:
  tools: tools/
  prompts: prompts/
  resources: resources/

3.将工具写入有组织的文件中

# tools/posts.py
from nextmcp import NextMCP

app = NextMCP.from_config()

@app.tool()
def create_post(title: str, content: str) -> dict:
    """Create a new blog post"""
    return {"id": 1, "title": title, "content": content}

@app.tool()
def list_posts() -> list:
    """List all blog posts"""
    return [{"id": 1, "title": "First Post"}]

4.单线服务器设置

# server.py
from nextmcp import NextMCP

# Auto-discovers all tools, prompts, and resources
app = NextMCP.from_config()

if __name__ == "__main__":
    app.run()

5.运行服务器

python server.py

就是这样！所有工具、提示和资源都会自动发现和注册。

手动方法

只需几个工具的简单项目。

1.创建新项目

mcp init my-bot
cd my-bot

2.写下你的第一个工具

# app.py
from nextmcp import NextMCP

app = NextMCP("my-bot")

@app.tool()
def greet(name: str) -> str:
    """Greet someone by name"""
    return f"Hello, {name}!"

if __name__ == "__main__":
    app.run()

3.运行服务器

mcp run app.py

您的MCP服务器现在正在运行 greet 工具可用。

基于惯例的项目结构

NextMCP v0.3.0引入了一个强大的基于约定的架构，其灵感来自Next.js的基于文件的路由。这种方法可以从目录结构中自动发现和注册工具、提示和资源，消除样板并改进项目组织。

为什么以公约为基础？

之前（手动注册）：

# app.py - 200+ lines of boilerplate
from nextmcp import NextMCP

app = NextMCP("my-server")

@app.tool()
def create_post(...):
    ...

@app.tool()
def update_post(...):
    ...

@app.tool()
def delete_post(...):
    ...

@app.prompt()
def writing_workflow(...):
    ...

@app.resource("blog://posts/recent")
def recent_posts():
    ...

# ... 20+ more primitives mixed together

之后（基于公约）：

# server.py - Just 3 lines!
from nextmcp import NextMCP

app = NextMCP.from_config()

if __name__ == "__main__":
    app.run()

项目结构

在标准目录中组织图元：

my-mcp-server/
├── nextmcp.config.yaml      # Project configuration
├── server.py                # Entry point
├── tools/                   # Tool definitions
│   ├── __init__.py
│   ├── posts.py            # Post management tools
│   └── comments.py         # Comment management tools
├── prompts/                 # Prompt templates
│   ├── __init__.py
│   └── workflows.py        # Workflow prompts
└── resources/               # Resource providers
    ├── __init__.py
    └── blog_resources.py   # Blog data resources

运作原理

1.汽车发现发动机

NextMCP扫描您的目录结构，并自动发现已装饰的函数：

# tools/posts.py
from nextmcp import NextMCP

app = NextMCP.from_config()

@app.tool()
def create_post(title: str, content: str) -> dict:
    """Create a new blog post"""
    return {"id": 1, "title": title}

@app.tool()
def list_posts(limit: int = 10) -> list:
    """List recent blog posts"""
    return []

发现引擎：

• 递归扫描 tools/, prompts/，以及 resources/ 目录
• 导入Python模块并检查修饰函数
• 自动注册所有发现的基元
• 跳跃 __init__.py 和 test_*.py 文件

2.配置文件

通过以下方式控制发现行为 nextmcp.config.yaml:

name: my-mcp-server
version: 1.0.0
description: My awesome MCP server

# Enable/disable auto-discovery
auto_discover: true

# Customize directory paths
discovery:
  tools: tools/
  prompts: prompts/
  resources: resources/

# Server configuration
server:
  host: 0.0.0.0
  port: 8000
  transport: stdio

# Middleware pipeline
middleware:
  - nextmcp.middleware.log_calls
  - nextmcp.middleware.error_handler

3.从配置加载

使用 from_config() 自动设置的类方法：

from nextmcp import NextMCP

# Load configuration and auto-discover primitives
app = NextMCP.from_config()

# Optional: specify custom config file or base path
app = NextMCP.from_config(
    config_file="custom.yaml",
    base_path="/path/to/project"
)

发现规则

自动发现引擎遵循以下规则：

1. 目录扫描：递归搜索已配置的目录
2. 模块导入：动态导入所有 .py 文件
3. 装饰检测：查找带有MCP装饰标记的函数
4. 自动配准：在应用程序中注册发现的基元
5. 文件排除：跳过 __init__.py 和 test_*.py 文件

组织大型项目

对于大型项目，使用子目录和模块：

tools/
├── __init__.py
├── posts/
│   ├── __init__.py
│   ├── create.py
│   ├── update.py
│   └── delete.py
├── comments/
│   ├── __init__.py
│   └── moderate.py
└── users/
    ├── __init__.py
    └── manage.py

子目录中的所有工具都会被自动发现。

验证

验证您的项目结构：

from nextmcp import validate_project_structure

# Check if project follows conventions
results = validate_project_structure()

if results["valid"]:
    print(f"✓ Found {results['stats']['tools']} tool files")
    print(f"✓ Found {results['stats']['prompts']} prompt files")
    print(f"✓ Found {results['stats']['resources']} resource files")
else:
    print("Errors:", results["errors"])
    print("Warnings:", results["warnings"])

益处

1. 关注点分离 -专用目录中的工具、提示和资源
2. 可扩展性 -通过创建文件添加新图元，无需注册
3. 团队协作 -为多个开发人员提供清晰的结构
4. 零沸点板 -无手动注册码
5. 类型安全 -具有组织模块的完整IDE支持
6. 测试 -易于单独测试单个模块

从手动注册迁移

现有的手动项目工作不变。要逐步迁移：

# You can mix both approaches!
from nextmcp import NextMCP

# Start with auto-discovery
app = NextMCP.from_config()

# Add manual tools as needed
@app.tool()
def legacy_tool():
    """This still works!"""
    return "result"

看 examples/blog_server/ 一个完整的基于惯例的项目。

身份验证和授权

NextMCP提供了一个全面的身份验证和授权系统，支持API密钥、JWT令牌、会话、OAuth 2.0和细粒度RBAC。该系统已投入生产，并提供了大量文档和示例。

为什么要对MCP进行身份验证？

MCP服务器通常需要：

• 保护敏感工具 未经授权的访问
• 实施基于角色的访问 （管理员、用户、查看器）
• 跟踪谁执行了操作 用于审核日志
• 与现有的身份验证系统集成 （API密钥、JWT、OAuth 2.0）
• 使用外部提供商对用户进行身份验证 （GitHub、谷歌等）

快速开始

API密钥验证

保护工具的最简单方法：

from nextmcp import NextMCP
from nextmcp.auth import APIKeyProvider, AuthContext, requires_auth_async

app = NextMCP("secure-server")

# Configure API key provider
api_key_provider = APIKeyProvider(
    valid_keys={
        "admin-key-123": {
            "user_id": "admin1",
            "username": "admin",
            "roles": ["admin"],
            "permissions": ["read:*", "write:*"],
        },
        "user-key-456": {
            "user_id": "user1",
            "username": "alice",
            "roles": ["user"],
            "permissions": ["read:posts"],
        }
    }
)

# Protected tool - requires authentication
@app.tool()
@requires_auth_async(provider=api_key_provider)
async def protected_tool(auth: AuthContext, data: str) -> dict:
    """Only authenticated users can access this."""
    return {
        "message": f"Hello {auth.username}",
        "data": data,
        "user_id": auth.user_id
    }

JWT令牌身份验证

对于基于无状态令牌的身份验证：

from nextmcp.auth import JWTProvider

# Configure JWT provider
jwt_provider = JWTProvider(
    secret_key="your-secret-key",
    algorithm="HS256",
    verify_exp=True
)

# Login endpoint that generates tokens
@app.tool()
async def login(username: str, password: str) -> dict:
    """Login and receive a JWT token."""
    # Validate credentials (check database, etc.)

    # Generate token
    token = jwt_provider.create_token(
        user_id=f"user_{username}",
        roles=["user"],
        permissions=["read:posts", "write:posts"],
        username=username,
        expires_in=3600  # 1 hour
    )

    return {"token": token, "expires_in": 3600}

# Use the token for authentication
@app.tool()
@requires_auth_async(provider=jwt_provider)
async def secure_action(auth: AuthContext) -> dict:
    """Requires valid JWT token."""
    return {"user": auth.username, "action": "performed"}

内置身份验证提供程序

NextMCP包括三个生产就绪的身份验证提供商：

基于角色的访问控制（RBAC）

根据用户角色控制访问：

from nextmcp.auth import requires_role_async

# Only admins can access this tool
@app.tool()
@requires_auth_async(provider=api_key_provider)
@requires_role_async("admin")
async def admin_tool(auth: AuthContext) -> dict:
    """Admin-only functionality."""
    return {"action": "admin action performed"}

# Users or admins can access
@app.tool()
@requires_auth_async(provider=api_key_provider)
@requires_role_async("user", "admin")  # Either role works
async def user_tool(auth: AuthContext) -> dict:
    """User or admin can access."""
    return {"action": "user action"}

基于权限的访问控制

具有特定权限的细粒度控制：

from nextmcp.auth import RBAC, requires_permission_async

# Set up RBAC system
rbac = RBAC()

# Define permissions
rbac.define_permission("read:posts", "Read blog posts")
rbac.define_permission("write:posts", "Create and edit posts")
rbac.define_permission("delete:posts", "Delete posts")

# Define roles with permissions
rbac.define_role("viewer", "Read-only access")
rbac.assign_permission_to_role("viewer", "read:posts")

rbac.define_role("editor", "Full content management")
rbac.assign_permission_to_role("editor", "read:posts")
rbac.assign_permission_to_role("editor", "write:posts")
rbac.assign_permission_to_role("editor", "delete:posts")

# Require specific permission
@app.tool()
@requires_auth_async(provider=api_key_provider)
@requires_permission_async("write:posts")
async def create_post(auth: AuthContext, title: str) -> dict:
    """Requires write:posts permission."""
    return {"status": "created", "title": title}

# Multiple permissions (user needs at least one)
@app.tool()
@requires_auth_async(provider=api_key_provider)
@requires_permission_async("admin:posts", "delete:posts")
async def delete_post(auth: AuthContext, post_id: int) -> dict:
    """Requires admin:posts OR delete:posts permission."""
    return {"status": "deleted", "post_id": post_id}

权限通配符

支持通配符权限：

# Admin with wildcard - matches ALL permissions
rbac.define_role("admin", "Full access")
rbac.assign_permission_to_role("admin", "*")

# Namespace wildcard - matches all admin permissions
rbac.assign_permission_to_role("moderator", "admin:*")

# moderator has: admin:users, admin:posts, admin:settings, etc.

AuthContext

这 AuthContext object作为第一个参数被注入到受保护的工具中：

@app.tool()
@requires_auth_async(provider=api_key_provider)
async def my_tool(auth: AuthContext, param: str) -> dict:
    # Access user information
    user_id = auth.user_id           # Unique user ID
    username = auth.username         # Human-readable name

    # Check roles and permissions
    is_admin = auth.has_role("admin")
    can_write = auth.has_permission("write:posts")

    # Access metadata
    department = auth.metadata.get("department")

    return {
        "user": username,
        "is_admin": is_admin,
        "can_write": can_write
    }

中间件堆叠

堆栈身份验证和授权装饰器：

@app.tool()                                           # 4. Register as tool
@requires_auth_async(provider=api_key_provider)      # 3. Authenticate user
@requires_role_async("admin")                        # 2. Check role
@requires_permission_async("delete:users")           # 1. Check permission (executes first)
async def delete_user(auth: AuthContext, user_id: int) -> dict:
    """Requires authentication, admin role, AND delete:users permission."""
    return {"status": "deleted", "user_id": user_id}

会话管理

使用SessionProvider进行基于会话的身份验证：

from nextmcp.auth import SessionProvider

session_provider = SessionProvider(session_timeout=3600)  # 1 hour

@app.tool()
async def login(username: str, password: str) -> dict:
    """Create a new session."""
    # Validate credentials...

    # Create session
    session_id = session_provider.create_session(
        user_id=f"user_{username}",
        username=username,
        roles=["user"],
        permissions=["read:posts"]
    )

    return {"session_id": session_id, "expires_in": 3600}

@app.tool()
async def logout(session_id: str) -> dict:
    """Destroy a session."""
    success = session_provider.destroy_session(session_id)
    return {"logged_out": success}

# Use session for authentication
@app.tool()
@requires_auth_async(provider=session_provider)
async def protected_tool(auth: AuthContext) -> dict:
    """Requires valid session."""
    return {"user": auth.username}

从配置加载RBAC

在配置中定义角色和权限：

from nextmcp.auth import RBAC

rbac = RBAC()

config = {
    "permissions": [
        {"name": "read:posts", "description": "Read posts"},
        {"name": "write:posts", "description": "Write posts"},
        {"name": "delete:posts", "description": "Delete posts"},
    ],
    "roles": [
        {
            "name": "viewer",
            "description": "Read-only",
            "permissions": ["read:posts"]
        },
        {
            "name": "editor",
            "description": "Full content management",
            "permissions": ["read:posts", "write:posts", "delete:posts"]
        }
    ]
}

rbac.load_from_config(config)

自定义身份验证提供程序

创建自己的身份验证提供程序：

from nextmcp.auth import AuthProvider, AuthResult, AuthContext

class CustomAuthProvider(AuthProvider):
    """Custom authentication using external service."""

    async def authenticate(self, credentials: dict) -> AuthResult:
        """Validate credentials against external service."""
        token = credentials.get("token")

        # Call your external auth service
        user_data = await external_auth_service.validate(token)

        if not user_data:
            return AuthResult.failure("Invalid token")

        # Build auth context
        context = AuthContext(
            authenticated=True,
            user_id=user_data["id"],
            username=user_data["name"],
        )

        # Add roles from external service
        for role in user_data.get("roles", []):
            context.add_role(role)

        return AuthResult.success_result(context)

错误处理

身份验证错误作为例外情况出现：

from nextmcp.auth import PermissionDeniedError
from nextmcp.auth.middleware import AuthenticationError

try:
    # Call protected tool without credentials
    result = await protected_tool(data="test")
except AuthenticationError as e:
    print(f"Auth failed: {e}")

try:
    # Call tool without required permission
    result = await admin_tool()
except PermissionDeniedError as e:
    print(f"Permission denied: {e}")
    print(f"Required: {e.required}")
    print(f"User: {e.user_id}")

安全最佳实践

1. 永远不要泄露秘密：使用环境变量作为密钥/秘密
2. 使用HTTPS/TLS：始终加密生产中的流量
3. 定期旋转按键实施关键的轮换政策
4. 短令牌过期：平衡安全性和用户体验（1-24小时）
5. 记录身份验证尝试：跟踪成功和失败的身份验证
6. 验证所有输入：从不信任客户提供的数据
7. 使用强大的秘密：生成 secrets.token_urlsafe(32)
8. 实施速率限制：防止暴力攻击

OAuth 2.0身份验证

NextMCP包括一个完整的OAuth 2.0实现，支持PKCE，GitHub和Google的即用型提供商，以及一个灵活的会话管理系统。

OAuth功能

• OAuth 2.0与PKCE：使用PKCE进行安全身份验证的完整授权码流
• 即用型提供商：GitHub和Google OAuth提供程序具有合理的默认值
• 会话管理：具有内存和基于文件的后端的持久令牌存储
• 认证元数据协议：服务器可以向MCP主机宣布身份验证要求
• 请求执行：运行时中间件验证令牌、作用域和权限
• 自动令牌刷新：内置令牌刷新处理
• 生产准备就绪：全面的测试和错误处理

快速开始使用Google OAuth

from fastmcp import FastMCP
from nextmcp.auth import GoogleOAuthProvider, create_auth_middleware
from nextmcp.session import FileSessionStore
from nextmcp.protocol import AuthRequirement, AuthMetadata, AuthFlowType

mcp = FastMCP("My Secure Server")

# Set up Google OAuth provider
google = GoogleOAuthProvider(
    client_id=os.getenv("GOOGLE_CLIENT_ID"),
    client_secret=os.getenv("GOOGLE_CLIENT_SECRET"),
    scopes=["openid", "email", "profile"]
)

# Create auth middleware with session storage
auth_middleware = create_auth_middleware(
    provider=google,
    requirement=AuthRequirement.REQUIRED,
    session_store=FileSessionStore("./sessions")
)

# Apply middleware to server
mcp.use(auth_middleware)

# Expose auth metadata so hosts know this server requires OAuth
auth_metadata = AuthMetadata(
    requirement=AuthRequirement.REQUIRED,
    flow_type=AuthFlowType.OAUTH,
    provider_name="google",
    scopes=["openid", "email", "profile"]
)
mcp.set_auth_metadata(auth_metadata)

# Tools now require authentication
@mcp.tool()
async def get_user_data(ctx: Context) -> str:
    return f"Hello {ctx.auth.username}!"

GitHub OAuth

from nextmcp.auth import GitHubOAuthProvider

github = GitHubOAuthProvider(
    client_id=os.getenv("GITHUB_CLIENT_ID"),
    client_secret=os.getenv("GITHUB_CLIENT_SECRET"),
    scopes=["user:email", "read:user"]
)

auth_middleware = create_auth_middleware(
    provider=github,
    requirement=AuthRequirement.REQUIRED,
    session_store=FileSessionStore("./sessions")
)

mcp.use(auth_middleware)

多提供商支持

在同一应用程序中使用多个OAuth提供程序：

from nextmcp.auth import GitHubOAuthProvider, GoogleOAuthProvider

# Set up both providers
github = GitHubOAuthProvider(...)
google = GoogleOAuthProvider(...)

# Different tools can use different providers
@mcp.tool()
@requires_auth_async(provider=github)
async def github_tool(ctx: Context) -> dict:
    return {"provider": "github", "user": ctx.auth.username}

@mcp.tool()
@requires_auth_async(provider=google)
async def google_tool(ctx: Context) -> dict:
    return {"provider": "google", "user": ctx.auth.username}

会话管理

会话存储持久化身份验证状态并处理令牌刷新：

from nextmcp.session import FileSessionStore, MemorySessionStore

# File-based session storage (persists across restarts)
file_store = FileSessionStore(
    directory="./sessions",
    auto_cleanup=True,
    cleanup_interval=3600  # Clean up expired sessions every hour
)

# In-memory session storage (for development)
memory_store = MemorySessionStore()

# Sessions automatically handle token refresh
auth_middleware = create_auth_middleware(
    provider=google,
    session_store=file_store,
    auto_refresh=True  # Automatically refresh expired tokens
)

OAuth范围和权限

在工具级别验证OAuth作用域：

from nextmcp.auth import requires_scope_async

@mcp.tool()
@requires_auth_async(provider=google)
@requires_scope_async("https://www.googleapis.com/auth/calendar")
async def access_calendar(ctx: Context) -> dict:
    """Requires Google Calendar scope."""
    return {"calendar": "data"}

认证元数据协议

auth元数据协议允许MCP服务器向主机应用程序（如Claude Desktop、Cursor等）宣布其身份验证要求：

from nextmcp.protocol import AuthMetadata, AuthRequirement, AuthFlowType

# Define authentication requirements
auth_metadata = AuthMetadata(
    requirement=AuthRequirement.REQUIRED,  # or OPTIONAL, NONE
    flow_type=AuthFlowType.OAUTH,         # or API_KEY, JWT
    provider_name="google",
    scopes=["openid", "email", "profile"],
    authorization_url="https://accounts.google.com/o/oauth2/v2/auth",
    token_url="https://oauth2.googleapis.com/token"
)

# Expose to hosts via server metadata
mcp.set_auth_metadata(auth_metadata)

主机可以查询此元数据：

• 确定是否需要身份验证
• 识别OAuth提供者和流类型
• 获取授权和令牌URL
• 请求适当的范围
• 引导用户进行身份验证

自定义OAuth提供程序

为其他服务创建自定义OAuth提供程序：

from nextmcp.auth import OAuthProvider, OAuthConfig

class CustomOAuthProvider(OAuthProvider):
    """Custom OAuth provider for your service."""

    def __init__(self, client_id: str, client_secret: str):
        config = OAuthConfig(
            client_id=client_id,
            client_secret=client_secret,
            authorization_url="https://your-service.com/oauth/authorize",
            token_url="https://your-service.com/oauth/token",
            scopes=["read", "write"],
            redirect_uri="http://localhost:8000/callback"
        )
        super().__init__(config)

    async def get_user_info(self, access_token: str) -> dict:
        """Fetch user information from your service."""
        async with aiohttp.ClientSession() as session:
            async with session.get(
                "https://your-service.com/api/user",
                headers={"Authorization": f"Bearer {access_token}"}
            ) as response:
                return await response.json()

测试OAuth流

使用提供的帮助程序以交互方式测试OAuth流：

# Run the OAuth token helper
python examples/auth/oauth_token_helper.py

# Follow the interactive prompts to:
# 1. Choose provider (GitHub or Google)
# 2. Get authorization URL
# 3. Complete OAuth flow
# 4. Receive access token
# 5. Test API calls with token

请参阅全面的OAuth文档：

• 建筑.md -系统设计和数据流
• OAUTH_TESTING_SETUP.md -设置OAuth凭据
• MIGRATION_GUIDE.md -向现有服务器添加身份验证
• HOST_INTEGRATION.md -MCP主机开发人员指南

示例

查看完整的身份验证示例：

• examples/auth/complete_oauth_server.py -具有会话管理的生产就绪Google OAuth
• examples/auth/github_oauth_server.py -GitHub OAuth身份验证示例
• examples/auth/google_oauth_server.py -Google OAuth身份验证示例
• examples/auth/multi_provider_server.py -一台服务器上有多个OAuth提供程序
• examples/auth/session_management_example.py -高级会话管理工作流
• examples/auth/combined_auth_server.py -OAuth与RBAC和权限相结合
• examples/auth/manifest_server.py -OAuth权限清单
• examples/auth/oauth_token_helper.py -交互式OAuth测试工具
• examples/auth_api_key/ -基于角色访问的API密钥身份验证
• examples/auth_jwt/ -使用登录端点进行JWT令牌身份验证
• examples/auth_rbac/ -具有细粒度权限的高级RBAC

核心概念

创建应用程序

from nextmcp import NextMCP

app = NextMCP(
    name="my-mcp-server",
    description="A custom MCP server"
)

注册工具

@app.tool()
def calculate(x: int, y: int) -> int:
    """Add two numbers"""
    return x + y

# With custom name and description
@app.tool(name="custom_name", description="A custom tool")
def my_function(data: str) -> dict:
    return {"result": data}

添加中间件

中间件封装您的工具以添加跨领域功能。

全局中间件（适用于所有工具）

from nextmcp import log_calls, error_handler

# Add middleware that applies to all tools
app.add_middleware(log_calls)
app.add_middleware(error_handler)

@app.tool()
def my_tool(x: int) -> int:
    return x * 2  # This will be logged and error-handled automatically

工具专用中间件

from nextmcp import cache_results, require_auth

@app.tool()
@cache_results(ttl_seconds=300)  # Cache for 5 minutes
def expensive_operation(param: str) -> dict:
    # Expensive computation here
    return {"result": perform_calculation(param)}

@app.tool()
@require_auth(valid_keys={"secret-key-123"})
def protected_tool(auth_key: str, data: str) -> str:
    return f"Protected: {data}"

内置中间件

NextMCP包括几个生产就绪的中间件：

• log_calls -记录所有工具调用并计时
• error_handler -捕获异常并返回结构化错误
• require_auth(valid_keys) -API密钥验证
• rate_limit(max_calls, time_window) -速率限制
• cache_results(ttl_seconds) -响应缓存
• `validate_inputs(validators)`** -自定义输入验证
• timeout(seconds) -执行超时

所有中间件也有异步变体（例如。， log_calls_async, error_handler_async等等）与异步工具一起使用。

异步支持

NextMCP完全支持异步/等待模式，允许您构建可以处理并发I/O操作的高性能工具。

基本异步工具

from nextmcp import NextMCP
import asyncio

app = NextMCP("async-app")

@app.tool()
async def fetch_data(url: str) -> dict:
    """Fetch data from an API asynchronously"""
    # Use async libraries like httpx, aiohttp, etc.
    await asyncio.sleep(0.1)  # Simulate API call
    return {"url": url, "data": "fetched"}

异步中间件

为异步工具使用异步中间件变体：

from nextmcp import log_calls_async, error_handler_async, cache_results_async

app.add_middleware(log_calls_async)
app.add_middleware(error_handler_async)

@app.tool()
@cache_results_async(ttl_seconds=300)
async def expensive_async_operation(param: str) -> dict:
    await asyncio.sleep(1)  # Simulate expensive operation
    return {"result": param}

并行操作

async的真正威力在于同时处理多个操作：

@app.tool()
async def fetch_multiple_sources(sources: list) -> dict:
    """Fetch data from multiple sources concurrently"""
    async def fetch_one(source: str):
        # Each fetch happens concurrently, not sequentially
        await asyncio.sleep(0.1)
        return {"source": source, "data": "..."}

    # Gather results concurrently - much faster than sequential!
    results = await asyncio.gather(*[fetch_one(s) for s in sources])
    return {"sources": results}

性能比较：

• 顺序：4个源×0.1s=0.4s
• 并发（异步）：~0.1秒（一次全部！）

混合同步和异步工具

您可以在同一个应用程序中同时使用同步和异步工具：

@app.tool()
def sync_tool(x: int) -> int:
    """Regular synchronous tool"""
    return x * 2

@app.tool()
async def async_tool(x: int) -> int:
    """Async tool for I/O operations"""
    await asyncio.sleep(0.1)
    return x * 3

何时使用异步

使用async用于：

• HTTP API调用（带有 httpx, aiohttp)
• 数据库查询（带 asyncpg, motor)
• 文件I/O操作
• 多个并发操作
• WebSocket连接

坚持同步：

• CPU密集型操作（大量计算）
• 无需I/O的简单操作
• 当第三方库不支持异步

看 examples/async_weather_bot/ 对于一个完整的异步示例。

使用Pydantic进行模式验证

from nextmcp import NextMCP
from pydantic import BaseModel

app = NextMCP("my-server")

class WeatherInput(BaseModel):
    city: str
    units: str = "fahrenheit"

@app.tool()
def get_weather(city: str, units: str = "fahrenheit") -> dict:
    # Input automatically validated against WeatherInput schema
    return {"city": city, "temp": 72, "units": units}

提示词

提示是用户驱动的工作流模板，用于指导人工智能交互。它们由用户明确调用（不是由AI自动调用），可以引用可用的工具和资源。

基本提示

from nextmcp import NextMCP

app = NextMCP("my-server")

@app.prompt()
def vacation_planner(destination: str, budget: int) -> str:
    """Plan a vacation itinerary."""
    return f"""
    Plan a vacation to {destination} with a budget of ${budget}.

    Use these tools:
    - flight_search: Find flights
    - hotel_search: Find accommodations

    Check these resources:
    - resource://user/preferences
    - resource://calendar/availability
    """

论点完成提示

from nextmcp import argument

@app.prompt(description="Research a topic", tags=["research"])
@argument("topic", description="What to research", suggestions=["Python", "MCP", "FastMCP"])
@argument("depth", suggestions=["basic", "detailed", "comprehensive"])
def research_prompt(topic: str, depth: str = "basic") -> str:
    """Generate a research prompt with the specified depth."""
    return f"Research {topic} at {depth} level..."

# Dynamic completion
@app.prompt_completion("research_prompt", "topic")
async def complete_topics(partial: str) -> list[str]:
    """Provide dynamic topic suggestions."""
    topics = await fetch_available_topics()
    return [t for t in topics if partial.lower() in t.lower()]

异步提示

@app.prompt(tags=["analysis"])
async def analyze_prompt(data_source: str) -> str:
    """Generate analysis prompt with real-time data."""
    data = await fetch_data(data_source)
    return f"Analyze this data: {data}"

何时使用提示：

• 指导复杂的多步骤工作流程
• 为常见任务提供模板
• 构建AI交互
• 参考可用的工具和资源

看 examples/knowledge_base/ 查看使用提示的完整示例。

资源

资源通过唯一的URI提供对上下文数据的只读访问。它们是应用程序驱动的，允许人工智能在不触发操作的情况下访问信息。

直接资源

from nextmcp import NextMCP

app = NextMCP("my-server")

@app.resource("file:///logs/app.log", description="Application logs")
def app_logs() -> str:
    """Provide access to application logs."""
    with open("/var/logs/app.log") as f:
        return f.read()

@app.resource("config://app/settings", mime_type="application/json")
def app_settings() -> dict:
    """Provide application configuration."""
    return {
        "theme": "dark",
        "language": "en",
        "max_results": 100
    }

资源模板

模板允许对动态资源进行参数化访问：

@app.resource_template("weather://forecast/{city}/{date}")
async def weather_forecast(city: str, date: str) -> dict:
    """Get weather forecast for a specific city and date."""
    return await fetch_weather(city, date)

@app.resource_template("file:///docs/{category}/{filename}")
def documentation(category: str, filename: str) -> str:
    """Access documentation files."""
    return Path(f"/docs/{category}/{filename}").read_text()

# Template parameter completion
@app.template_completion("weather_forecast", "city")
def complete_cities(partial: str) -> list[str]:
    """Suggest city names."""
    return ["London", "Paris", "Tokyo", "New York"]

可订阅资源

资源可以在更改时通知订阅者：

@app.resource(
    "config://live/settings",
    subscribable=True,
    max_subscribers=50
)
async def live_settings() -> dict:
    """Provide live configuration that can change."""
    return await load_live_config()

# Notify subscribers when config changes
app.notify_resource_changed("config://live/settings")

# Manage subscriptions
app.subscribe_to_resource("config://live/settings", "subscriber_id")
app.unsubscribe_from_resource("config://live/settings", "subscriber_id")

异步资源

@app.resource("db://users/recent")
async def recent_users() -> list[dict]:
    """Get recently active users from database."""
    return await db.query("SELECT * FROM users ORDER BY last_active DESC LIMIT 10")

何时使用资源：

• 提供只读数据访问
• 显示配置和设置
• 共享应用程序状态
• 提供实时数据馈送（带订阅）

资源URI可以使用任何方案：

• file:// -文件系统访问
• config:// -配置数据
• db:// -数据库查询
• api:// -API外部数据
• 为您的用例定制方案

看 examples/knowledge_base/ 查看使用资源和模板的完整示例。

配置

NextMCP支持自动合并多个配置源：

from nextmcp import load_config

# Load from config.yaml and .env
config = load_config(config_file="config.yaml")

# Access configuration
host = config.get_host()
port = config.get_port()
debug = config.is_debug()

# Custom config values
api_key = config.get("api_key", default="default-key")

config.yaml:

host: "0.0.0.0"
port: 8080
log_level: "DEBUG"
api_key: "my-secret-key"

.env:

MCP_HOST=0.0.0.0
MCP_PORT=8080
API_KEY=my-secret-key

WebSocket传输

NextMCP支持WebSocket传输，用于实时双向通信，非常适合聊天应用程序、实时更新和交互式工具。

服务器设置

from nextmcp import NextMCP
from nextmcp.transport import WebSocketTransport

app = NextMCP("websocket-server")

@app.tool()
async def send_message(username: str, message: str) -> dict:
    return {
        "status": "sent",
        "username": username,
        "message": message
    }

# Create WebSocket transport
transport = WebSocketTransport(app)

# Run on ws://localhost:8765
transport.run(host="0.0.0.0", port=8765)

客户端使用情况

from nextmcp.transport import WebSocketClient

async def main():
    async with WebSocketClient("ws://localhost:8765") as client:
        # List available tools
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

        # Invoke a tool
        result = await client.invoke_tool(
            "send_message",
            {"username": "Alice", "message": "Hello!"}
        )
        print(f"Result: {result}")

WebSocket功能

• 实时通信：低延迟的持久连接
• 双向：服务器可以向客户端推送更新
• JSON-RPC协议：工具调用的干净消息格式
• 多个客户端：处理多个并发连接
• 异步本机：基于Python的async/await实现高性能

何时使用WebSocket与HTTP

看 examples/websocket_chat/ 一个完整的WebSocket应用程序。

插件系统

NextMCP具有强大的插件系统，允许您通过模块化、可重用的组件扩展功能。

什么是插件？

插件是自包含的模块，可以：

• 在您的应用程序中注册新工具
• 添加中间件以解决跨领域问题
• 扩展核心功能
• 易于在项目之间共享和重用

创建插件

from nextmcp import Plugin

class MathPlugin(Plugin):
    name = "math-plugin"
    version = "1.0.0"
    description = "Mathematical operations"
    author = "Your Name"

    def on_load(self, app):
        @app.tool()
        def add(a: float, b: float) -> float:
            """Add two numbers"""
            return a + b

        @app.tool()
        def multiply(a: float, b: float) -> float:
            """Multiply two numbers"""
            return a * b

使用插件

方法1：自动发现

from nextmcp import NextMCP

app = NextMCP("my-app")

# Discover all plugins in a directory
app.discover_plugins("./plugins")

# Load all discovered plugins
app.load_plugins()

方法2：直接加载

from nextmcp import NextMCP
from my_plugins import MathPlugin

app = NextMCP("my-app")

# Load a specific plugin
app.use_plugin(MathPlugin)

插件生命周期

插件有三个生命周期挂钩：

1. on_init() -在插件初始化期间调用
2. on_load(app) -加载插件时调用（在此处注册工具）
3. \_卸载（） -卸载插件时调用（清理）

class LifecyclePlugin(Plugin):
    name = "lifecycle-example"
    version = "1.0.0"

    def on_init(self):
        # Early initialization
        self.config = {}

    def on_load(self, app):
        # Register tools and middleware
        @app.tool()
        def my_tool():
            return "result"

    def on_unload(self):
        # Cleanup resources
        self.config.clear()

带有中间件的插件

class TimingPlugin(Plugin):
    name = "timing"
    version = "1.0.0"

    def on_load(self, app):
        import time

        def timing_middleware(fn):
            def wrapper(*args, **kwargs):
                start = time.time()
                result = fn(*args, **kwargs)
                elapsed = (time.time() - start) * 1000
                print(f"⏱️ {fn.__name__} took {elapsed:.2f}ms")
                return result
            return wrapper

        app.add_middleware(timing_middleware)

插件依赖关系

插件可以声明对其他插件的依赖关系：

class DependentPlugin(Plugin):
    name = "advanced-math"
    version = "1.0.0"
    dependencies = ["math-plugin"]  # Loads math-plugin first

    def on_load(self, app):
        @app.tool()
        def factorial(n: int) -> int:
            # Can use tools from math-plugin
            return 1 if n <= 1 else n * factorial(n - 1)

管理插件

# List all loaded plugins
for plugin in app.plugins.list_plugins():
    print(f"{plugin['name']} v{plugin['version']} - {plugin['loaded']}")

# Get a specific plugin
plugin = app.plugins.get_plugin("math-plugin")

# Unload a plugin
app.plugins.unload_plugin("math-plugin")

# Check if plugin is loaded
if "math-plugin" in app.plugins:
    print("Math plugin is available")

插件最佳实践

1. 使用描述性名称 -使插件名称清晰唯一
2. 语义版本 -遵循semver（主要。次要。补丁）
3. 彻底记录 -添加描述和文档字符串
4. 优雅地处理错误 -在生命周期钩子中捕获异常
5. 声明依赖关系 -明确列出所需的插件
6. 实施清理 -使用 on_unload() 释放资源

看 examples/plugin_example/ 获取具有多种插件类型的完整插件演示。

度量与监控

NextMCP包括一个内置的指标系统，用于监控生产中的MCP应用程序。

快速开始

from nextmcp import NextMCP

app = NextMCP("my-app")
app.enable_metrics()  # That's it! Automatic metrics collection

@app.tool()
def my_tool():
    return "result"

自动度量

启用指标后，NextMCP会自动跟踪：

• tool_invocations_total -工具调用总数
• tool_duration_seconds -工具执行时间柱状图
• tool_completed_total -按状态（成功/错误）列出的已完成调用
• tool_errors_total -按错误类型分类的错误
• tool_active_invocations -当前正在执行的工具

所有指标都包括工具名称的标签和您配置的任何全局标签。

自定义指标

添加您自己的业务逻辑指标：

@app.tool()
def process_order(order_id: int):
    # Custom counter
    app.metrics.inc_counter("orders_processed")

    # Custom gauge
    app.metrics.set_gauge("current_queue_size", get_queue_size())

    # Custom histogram with timer
    with app.metrics.time_histogram("processing_duration"):
        result = process(order_id)

    return result

度量类型

计数器

单调递增的价值。用于：计数、总计。

counter = app.metrics.counter("requests_total")
counter.inc()  # Increment by 1
counter.inc(5)  # Increment by 5

测量

值可以上升或下降。用于：当前值、温度、队列大小。

gauge = app.metrics.gauge("active_connections")
gauge.set(10)  # Set to specific value
gauge.inc()    # Increment
gauge.dec()    # Decrement

直方图

价值观的分布。用于：持续时间、大小。

histogram = app.metrics.histogram("request_duration_seconds")
histogram.observe(0.25)

# Or use as timer
with app.metrics.time_histogram("duration"):
    # Code to time
    pass

导出指标

Prometheus格式

# Get metrics in Prometheus format
prometheus_data = app.get_metrics_prometheus()
print(prometheus_data)

输出：

# HELP my-app_tool_invocations_total Total tool invocations
# TYPE my-app_tool_invocations_total counter
my-app_tool_invocations_total{tool="my_tool"} 42.0

# HELP my-app_tool_duration_seconds Tool execution duration
# TYPE my-app_tool_duration_seconds histogram
my-app_tool_duration_seconds_bucket{tool="my_tool",le="0.005"} 10
my-app_tool_duration_seconds_bucket{tool="my_tool",le="0.01"} 25
my-app_tool_duration_seconds_sum{tool="my_tool"} 1.234
my-app_tool_duration_seconds_count{tool="my_tool"} 42

JSON格式

# Get metrics as JSON
json_data = app.get_metrics_json(pretty=True)

配置

app.enable_metrics(
    collect_tool_metrics=True,      # Track tool invocations
    collect_system_metrics=False,   # Track CPU/memory (future)
    collect_transport_metrics=False, # Track WebSocket/HTTP (future)
    labels={"env": "prod", "region": "us-west"}  # Global labels
)

带标签的指标

标签允许您对指标进行切片和切块：

counter = app.metrics.counter(
    "api_requests",
    labels={"method": "GET", "endpoint": "/users"}
)
counter.inc()

与监控系统集成

Prometheus格式与以下格式兼容：

• Prometheus用于刮擦和储存
• Grafana用于可视化
• AlertManager用于发出警报
• 任何兼容Prometheus的系统

看 examples/metrics_example/ 以获得完整的指标演示。

生产部署

NextMCP为使用Docker、云平台和Kubernetes将MCP服务器部署到生产环境提供了全面的工具。

快速开始

为您的项目生成Docker部署文件：

cd my-mcp-project
mcp init --docker --with-database

这将创建：

• Dockerfile -优化的多级构建（\<100MB）
• docker-compose.yml -完善的本地开发环境
• .dockerignore -最小Docker上下文

只需一个命令即可部署：

mcp deploy --platform docker

部署功能

• 健康检查 -Kubernetes兼容的活性和就绪性探针
• 优雅地关闭 -使用SIGTERM/SIGINT处理进行清洁端接
• 多阶段建筑 -以最小的大小优化Docker镜像
• 平台支持 -Docker、Railway、Render、Fly.io、Kubernetes
• 自动检测 -自动检测和配置依赖关系
• 安全加固 -非root用户，攻击面最小

健康检查

内置健康检查系统，用于监控应用程序的健康状况：

from nextmcp import NextMCP
from nextmcp.deployment import HealthCheck

app = NextMCP("my-app")
health = HealthCheck()

# Add custom readiness check
def check_database():
    return db.is_connected()

health.add_readiness_check("database", check_database)

# Health endpoints automatically available:
# GET /health        - Liveness probe
# GET /health/ready  - Readiness probe

健康检查类型：

• 活性：应用程序正在运行吗？（失败后重新启动）
• 准备就绪：应用程序是否已准备好为流量服务？（如果失败，请从负载平衡器中删除）

状态类型：

• healthy -所有检查均已通过
• unhealthy -一个或多个检查失败
• degraded -部分功能可用

优雅地关闭

干净地处理关机信号以防止数据丢失：

from nextmcp.deployment import GracefulShutdown

shutdown = GracefulShutdown(timeout=30.0)

# Add cleanup handlers
def cleanup_resources():
    db.close()
    cache.flush()

shutdown.add_cleanup_handler(cleanup_resources)
shutdown.register()

# Handles SIGTERM/SIGINT automatically
# Waits for in-flight requests to complete
# Runs cleanup handlers in order

Docker部署

生成Docker文件

# Basic Docker setup
mcp init --docker

# With PostgreSQL
mcp init --docker --with-database

# With PostgreSQL and Redis
mcp init --docker --with-database --with-redis

# Custom port
mcp init --docker --port 9000

本地部署

# Build and start
docker compose up --build

# View logs
docker compose logs -f

# Stop
docker compose down

Dockerfile功能

生成的Dockerfile包括：

• 多阶段构建 -独立的构建器和运行时阶段
• Python 3.10精简版 -最小基础映像（总计约100MB）
• 非root用户 -以用户身份运行 nextmcp （UID 1000）
• 健康检查 -内置HTTP健康检查
• 优化图层 -缓存依赖关系以实现更快的构建

云平台部署（测试版）

只需一个命令即可部署到流行的云平台。 注： 这些集成处于测试阶段，使用平台的CLI工具。欢迎社区测试和反馈！

铁路（贝塔）

mcp deploy --platform railway

特征：

• 自动HTTPS
• 环境变量
• 自动缩放
• 内置监控

要求： 安装 铁路CLI (npm install -g @railway/cli)

渲染（测试版）

mcp deploy --platform render

特征：

• 基于Git的部署
• 自动SSL
• 免费套餐可用
• 持久卷

要求： 安装 渲染CLI

Fly.io（测试版）

mcp deploy --platform fly

特征：

• 边缘部署
• 全球性分布
• WebSocket支持
• 自定义域名

要求： 安装 Fly CLI (flyctl)

Beta通知： 云平台部署使用各自的CLI工具。我们验证了命令的调用是否正确，但尚未在这些平台上测试实际部署。如果您成功使用这些（或遇到问题），请 打开一个问题 帮助提高集成度！

Kubernetes部署

健康检查已为Kubernetes做好准备：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: nextmcp-app
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: nextmcp
        image: my-nextmcp-app:latest
        ports:
        - containerPort: 8000
        # Liveness probe
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
        # Readiness probe
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"

环境配置

为不同的环境配置应用程序：

# .env file (development)
PORT=8000
LOG_LEVEL=DEBUG
DATABASE_URL=postgresql://localhost:5432/dev

# .env.production
PORT=8000
LOG_LEVEL=INFO
DATABASE_URL=postgresql://prod-db:5432/app

在应用程序中加载配置：

import os

app = NextMCP("my-app")

# Environment-specific configuration
port = int(os.getenv("PORT", "8000"))
log_level = os.getenv("LOG_LEVEL", "INFO")

app.run(host="0.0.0.0", port=port)

生产检查表

在部署到生产环境之前：

• \[\]配置并测试了健康检查
• \[\]已启用优雅关机
• \[\]保护环境变量（使用机密管理）
• \[\]数据库迁移正在运行
• \[\]已配置SSL/TLS证书
• \[\]监控和警报设置
• \[\]已配置日志聚合
• \[\]已实施备份策略
• \[\]已进行负载测试
• \[\]设置资源限制（CPU、内存）
• \[\]安全扫描已完成
• \[\]文件已更新

部署示例

查看完整的部署示例：

• examples/deployment_simple/ -带健康检查的基本部署

- 简单的健康和准备检查 - 优雅关闭 - 生产测井 - Docker部署就绪

• examples/deployment_docker/ -生产就绪部署

- 数据库与健康检查集成 - 指标收集 - 高级健康检查（磁盘空间、数据库） - 多服务Docker Compose - 环境配置 - 生产最佳实践

平台支持矩阵

测试状态：

• ✅ 全力支持：CI中的全面自动化测试，生产就绪
• 🧪 贝塔：CLI集成有效，但未在实际平台上进行测试-需要社区测试
• 🔄 计划的：尚未实施

Beta平台注意事项： Railway、Render和Fly.io部署使用各自的CLI工具。我们测试命令是否被正确调用，但完整的平台集成需要手动验证。如果您成功部署到这些平台，请在 !

故障排除

容器无法启动：

# Check logs
docker compose logs nextmcp-app

# Verify configuration
docker compose config

# Check port conflicts
lsof -i :8000

健康检查失败：

# Manual health check
curl http://localhost:8000/health

# Check container health
docker inspect nextmcp-app | grep Health -A 10

# View detailed logs
docker compose logs -f --tail=100

数据库连接问题：

# Test database connectivity
docker compose exec postgres pg_isready

# Check database logs
docker compose logs postgres

有关更多部署指南和故障排除，请参阅中的示例README examples/deployment_simple/ 和 examples/deployment_docker/.

CLI命令

NextMCP为常见的开发任务提供了丰富的CLI。

初始化新项目

# Create from template
mcp init my-project
mcp init my-project --template weather_bot
mcp init my-project --path /custom/path

# Generate Docker deployment files
mcp init --docker
mcp init --docker --with-database
mcp init --docker --with-database --with-redis
mcp init --docker --port 9000

部署到生产环境

# Auto-detect platform and deploy
mcp deploy

# Deploy to specific platform
mcp deploy --platform docker
mcp deploy --platform railway
mcp deploy --platform render
mcp deploy --platform fly

# Deploy without building
mcp deploy --platform docker --no-build

运行服务器

mcp run app.py
mcp run app.py --host 0.0.0.0 --port 8080
mcp run app.py --reload  # Auto-reload on changes

生成文档

mcp docs app.py
mcp docs app.py --output docs.md
mcp docs app.py --format json

生成manifest.json

生成一个清单文件，描述服务器的功能、工具、提示和资源：

# Print manifest to stdout
mcp manifest app.py

# Save to file
mcp manifest app.py --output manifest.json
mcp manifest app.py --save  # Shorthand for --output manifest.json

# Validate manifest
mcp manifest app.py --validate

生成的清单包括：

• 服务器元数据（名称、版本、描述）
• 能力声明（工具、提示、资源、日志记录、完成）
• 使用JSON Schema参数完成工具列表
• 带有参数规范的提示模板
• 具有URI模式的资源和资源模板
• 生成元数据（时间戳、自动发现信息、中间件、部署设置）

您还可以通过编程生成清单：

from nextmcp import NextMCP

app = NextMCP.from_config()

# Generate and save
manifest = app.generate_manifest("manifest.json")

# Or just generate without saving
manifest = app.generate_manifest()

验证清单安全性

使用静态分析验证安全问题清单：

# Validate a manifest file
mcp validate manifest.json

# Generate and validate from app
mcp validate --app app.py

# Fail on different risk levels
mcp validate manifest.json --fail-on high    # Blocks HIGH and CRITICAL
mcp validate manifest.json --fail-on medium  # Blocks MEDIUM, HIGH, and CRITICAL

# JSON output for CI/CD integration
mcp validate manifest.json --json

⚠️ 关键安全警告

清单验证不足以保证安全！

验证器执行静态分析以捕捉明显的问题，但 不能:

• ❌ 检测服务器实现中的恶意代码
• ❌ 验证身份验证/授权是否正确实施
• ❌ 检测运行时漏洞或业务逻辑缺陷
• ❌ 防止来自坚定对手的复杂攻击
• ❌ 即使验证通过，也要保证您的服务器是安全的

舱单可以伪造或损坏：

• 攻击者可以创建看似安全但隐藏恶意操作的虚假清单
• 清单可以声明代码中不存在的严格验证
• 清单中的架构可能与实际服务器行为不匹配
• 工具可以完全隐藏在清单中

验证器检查的内容：

• ✅ 危险的操作模式（删除、执行、管理命令）
• ✅ 缺少输入验证（无界字符串、无约束对象）
• ✅ 常见注入风险（SQL、命令、路径遍历、SSRF）
• ✅ 敏感数据暴露指标
• ✅ 大攻击面（许多暴露的工具）
• ✅ 缺少危险操作的身份验证指示器

将验证作为深度防御的一层：

Security Layer 1: Manifest Validation (this tool) ← Catches obvious issues
Security Layer 2: Static Code Analysis (Bandit, Semgrep) ← Finds vulnerabilities in code
Security Layer 3: Dependency Scanning (Snyk, Safety) ← Detects known CVEs
Security Layer 4: Manual Code Review ← Human security review
Security Layer 5: Penetration Testing ← Test for exploits
Security Layer 6: Runtime Monitoring ← Detect anomalies in production

最佳实践：

1. 永远不要只相信自己 -始终查看服务器代码
2. 纵深防御 -使用多个安全层
3. 最小权限原则 -仅公开必要的操作
4. 假设违约 -添加审计日志记录、速率限制、监控
5. 定期更新 -对每次更改进行重新验证

看 examples/security_validation/ 了解安全服务器与不安全服务器的详细示例。

显示版本

mcp version

示例

看看 examples/ 完整工作示例目录：

• 博客服务器 -基于惯例的项目结构，具有自动发现功能（5个工具、3个提示、4个资源）
• 安全验证 -显示安全服务器与不安全服务器的清单验证示例
• auth/ -OAuth 2.0身份验证示例：

- complete_auth_server.py -具有会话管理的生产就绪Google OAuth - github _ auth_server.py -GitHub OAuth身份验证 - google_oauth_server.py -Google OAuth身份验证 - 多提供者服务器.py -一台服务器上有多个OAuth提供程序 - 会话_管理_示例.py -高级会话管理 - combined_auth_server.py -具有RBAC和权限的OAuth - manifest_server.py -OAuth权限清单 - oauth_token_helper.py -交互式OAuth测试工具

• auth_api_key -基于角色访问控制的API密钥身份验证
• auth_jwt -具有登录端点和令牌生成的JWT令牌身份验证
• auth_rbac -高级RBAC，具有细粒度权限和通配符
• 部署_简单 -基本生产部署，包括健康检查和优雅关机
• 部署锁定器 -具有数据库、指标和Docker Compose的生产就绪部署
• 天气预报 -具有多种工具的天气信息服务器
• async_weather_bot -异步版本演示并发操作和异步中间件
• Websocket聊天 -使用WebSocket传输的实时聊天服务器
• 插件示例 -具有多种插件类型的插件系统演示
• 度量_示例 -使用自动和自定义指标进行指标和监控演示

发展

为发展而设立

# Clone the repository
git clone https://github.com/KeshavVarad/NextMCP.git
cd nextmcp

# Install in editable mode with dev dependencies
pip install -e ".[dev]"

# Install git pre-commit hooks (recommended)
./scripts/install-hooks.sh

# Run tests
pytest

# Run tests with coverage
pytest --cov=nextmcp --cov-report=html

# Format code
black nextmcp tests

# Lint code
ruff check nextmcp tests

# Type check
mypy nextmcp

预提交钩子

该存储库包含一个预提交钩子，在每次提交之前自动运行：

• 使用ruff检查并自动修复代码
• 黑色格式化代码
• 运行所有测试

使用以下工具安装挂钩：

./scripts/install-hooks.sh

钩子确保所有提交都通过linting和测试，防止CI失败。要绕过钩子（不推荐），请使用：

git commit --no-verify

运行测试

# Run all tests
pytest

# Run specific test file
pytest tests/test_core.py

# Run with verbose output
pytest -v

# Run with coverage
pytest --cov=nextmcp

建筑

NextMCP分为几个模块：

• core.py -Main NextMCP 类、应用程序生命周期和 from_config() 方法
• discovery.py -基于惯例的项目结构自动发现引擎
• tools.py -工具注册、元数据和文档生成
• middleware.py -用于常见用例的内置中间件
• config.py -配置管理（YAML、.env、环境变量）
• cli.py -基于Typer的CLI命令
• logging.py -集中式日志设置和实用程序

与FastMCP的比较

NextMCP以FastMCP为基础，提供：

路线图

完成

• \[x\] v0.1.0 -带工具原语的核心MCP服务器
• \[x\] v0.2.0版本 -完整的MCP图元（提示、资源、资源模板、订阅）
• \[x\] v0.3.0 -基于约定的架构（自动发现， from_config()，项目结构）
• \[x\] v0.4.0 -身份验证和授权（API密钥、JWT、会话、RBAC）
• \[x\] v0.5.0 -生产部署（健康检查、优雅关机、Docker、云平台）
• \[x\] v0.6.0 -OAuth 2.0身份验证（GitHub/Google提供商、PKCE、会话管理、Auth元数据协议）
• \[x\] 异步工具支持
• \[x\] WebSocket传输
• \[x\] 插件系统
• \[x\] 内置监控和指标

进行中

• \[\]MCP服务器注册和发现
• \[\]更多示例项目
• \[\]文档网站

计划的

v0.6.0-开发人员体验

• 热重新加载：具有自动文件监视功能的开发模式
• 增强的CLI: mcp dev, mcp validate, mcp test 命令
• 交互式调试：MCP工具的内置调试器

v0.7.0-高级部署

• 无服务器支持：AWS Lambda、谷歌云功能、Azure功能
• Kubernetes Helm图表：生产就绪的K8部署
• 包装分发: mcp package 适用于Docker、PyPI和无服务器

贡献

欢迎投稿！请随时提交拉取请求。

1. 分叉存储库
2. 创建功能分支(git checkout -b feature/amazing-feature)
3. 提交您的更改(git commit -m 'Add some amazing feature')
4. 推到分支(git push origin feature/amazing-feature)
5. 打开拉取请求

许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

致谢

• 建在上面 FastMCP
• 灵感来自 Next.js 开发者体验
• CLI由 类型

支持

• GitHub问题：
• 文档：\[即将发布\]

______________________________________________________________________

由以下材料制成❤️ NextMCP社区