nextmcp 0.6.0
Production-grade MCP server toolkit with minimal boilerplate
License: MIT
Requires Python: >=3.8
Keywords: mcp, server, sdk, fastmcp, framework

# NextMCP

[![Tests](https://github.com/KeshavVarad/NextMCP/workflows/Tests/badge.svg)](https://github.com/KeshavVarad/NextMCP/actions)
[![PyPI version](https://badge.fury.io/py/nextmcp.svg)](https://badge.fury.io/py/nextmcp)
[![Python versions](https://img.shields.io/pypi/pyversions/nextmcp.svg)](https://pypi.org/project/nextmcp/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)

**Production-grade MCP server toolkit with minimal boilerplate**

NextMCP is a Python SDK built on top of FastMCP that provides a developer-friendly experience for building MCP (Model Context Protocol) servers. Inspired by Next.js, it offers minimal setup, powerful middleware, and a rich CLI for rapid development.

## Features

- **Full MCP Specification** - Complete support for Tools, Prompts, and Resources primitives
- **Convention-Based Structure** - Next.js-inspired file-based organization with auto-discovery
- **Zero-Config Setup** - Single-line `NextMCP.from_config()` for instant project setup
- **Auto-Discovery** - Automatically discover and register primitives from directory structure
- **Production Deployment** - One-command deploy to Docker, Railway, Render, Fly.io with health checks
- **Health Checks** - Built-in liveness and readiness probes for Kubernetes and cloud platforms
- **Graceful Shutdown** - Clean termination with SIGTERM/SIGINT handling and resource cleanup
- **Minimal Boilerplate** - Get started with just a few lines of code
- **Decorator-based API** - Register tools, prompts, and resources with simple decorators
- **Async Support** - Full support for async/await across all primitives
- **Argument Completion** - Smart suggestions for prompt arguments and resource templates
- **Resource Subscriptions** - Real-time notifications when resources change
- **WebSocket Transport** - Real-time bidirectional communication for interactive applications
- **Authentication & Authorization** - Built-in support for API keys, JWT, sessions, OAuth 2.0, and RBAC
- **OAuth 2.0 Support** - Complete OAuth implementation with GitHub and Google providers, PKCE, session management
- **Global & Primitive-specific Middleware** - Add logging, auth, rate limiting, caching, and more
- **Rich CLI** - Scaffold projects, run servers, deploy, and generate docs with `mcp` commands
- **Configuration Management** - Support for `.env`, YAML config files, and environment variables
- **Schema Validation** - Optional Pydantic integration for type-safe inputs
- **Metrics & Monitoring** - Prometheus-compatible metrics with custom metric support
- **Production Ready** - Built-in error handling, logging, and comprehensive testing

## Installation

### Basic Installation

```bash
pip install nextmcp
```

### With Optional Dependencies

```bash
# 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]
```

## Quick Start

NextMCP offers two approaches: **Convention-Based** (recommended) for scalable projects, and **Manual** for simple use cases.

### Convention-Based Approach (Recommended)

Perfect for projects with multiple tools, prompts, and resources. Uses file-based organization for automatic discovery.

#### 1. Create project structure

```bash
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. Configure your project

```yaml
# 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. Write tools in organized files

```python
# 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. Single-line server setup

```python
# server.py
from nextmcp import NextMCP

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

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

#### 5. Run your server

```bash
python server.py
```

That's it! All tools, prompts, and resources are automatically discovered and registered.

### Manual Approach

For simple projects with just a few tools.

#### 1. Create a new project

```bash
mcp init my-bot
cd my-bot
```

#### 2. Write your first tool

```python
# 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. Run your server

```bash
mcp run app.py
```

Your MCP server is now running with the `greet` tool available.

## Convention-Based Project Structure

NextMCP v0.3.0 introduces a powerful convention-based architecture inspired by Next.js's file-based routing. This approach enables automatic discovery and registration of tools, prompts, and resources from your directory structure, eliminating boilerplate and improving project organization.

### Why Convention-Based?

**Before (Manual Registration):**
```python
# 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
```

**After (Convention-Based):**
```python
# server.py - Just 3 lines!
from nextmcp import NextMCP

app = NextMCP.from_config()

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

### Project Structure

Organize your primitives in standard directories:

```
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
```

### How It Works

#### 1. Auto-Discovery Engine

NextMCP scans your directory structure and automatically discovers decorated functions:

```python
# 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 []
```

The discovery engine:
- Recursively scans `tools/`, `prompts/`, and `resources/` directories
- Imports Python modules and inspects decorated functions
- Automatically registers all discovered primitives
- Skips `__init__.py` and `test_*.py` files

#### 2. Configuration File

Control discovery behavior with `nextmcp.config.yaml`:

```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. Loading from Config

Use the `from_config()` class method for automatic setup:

```python
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"
)
```

### Discovery Rules

The auto-discovery engine follows these rules:

1. **Directory Scanning**: Recursively searches configured directories
2. **Module Importing**: Dynamically imports all `.py` files
3. **Decorator Detection**: Finds functions with MCP decorator markers
4. **Automatic Registration**: Registers discovered primitives with the app
5. **File Exclusions**: Skips `__init__.py` and `test_*.py` files

### Organizing Large Projects

For large projects, use subdirectories and modules:

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

All tools in subdirectories are automatically discovered.

### Validation

Validate your project structure:

```python
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"])
```

### Benefits

1. **Separation of Concerns** - Tools, prompts, and resources in dedicated directories
2. **Scalability** - Add new primitives by creating files, no registration needed
3. **Team Collaboration** - Clear structure for multiple developers
4. **Zero Boilerplate** - No manual registration code
5. **Type Safety** - Full IDE support with organized modules
6. **Testing** - Easy to test individual modules in isolation

### Migrating from Manual Registration

Existing manual projects work unchanged. To migrate gradually:

```python
# 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"
```

See `examples/blog_server/` for a complete convention-based project.

## Authentication & Authorization

NextMCP provides a comprehensive authentication and authorization system with support for API keys, JWT tokens, sessions, OAuth 2.0, and fine-grained RBAC. The system is production-ready with extensive documentation and examples.

### Why Authentication for MCP?

MCP servers often need to:
- **Protect sensitive tools** from unauthorized access
- **Implement role-based access** (admin, user, viewer)
- **Track who performed actions** for audit logs
- **Integrate with existing auth systems** (API keys, JWT, OAuth 2.0)
- **Authenticate users with external providers** (GitHub, Google, etc.)

### Quick Start

#### API Key Authentication

The simplest way to protect your tools:

```python
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 Token Authentication

For stateless token-based auth:

```python
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"}
```

### Built-in Auth Providers

NextMCP includes three production-ready authentication providers:

| Provider | Use Case | Features |
|----------|----------|----------|
| **APIKeyProvider** | Simple API key auth | Pre-configured keys, custom validators, secure generation |
| **JWTProvider** | Token-based auth | Automatic expiration, signature verification, stateless |
| **SessionProvider** | Session-based auth | In-memory sessions, automatic cleanup, session management |

### Role-Based Access Control (RBAC)

Control access based on user roles:

```python
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"}
```

### Permission-Based Access Control

Fine-grained control with specific permissions:

```python
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}
```

### Permission Wildcards

Support for wildcard permissions:

```python
# 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

The `AuthContext` object is injected as the first parameter to protected tools:

```python
@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
    }
```

### Middleware Stacking

Stack authentication and authorization decorators:

```python
@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}
```

### Session Management

Using the SessionProvider for session-based authentication:

```python
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}
```

### Loading RBAC from Configuration

Define roles and permissions in configuration:

```python
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)
```

### Custom Auth Providers

Create your own authentication provider:

```python
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)
```

### Error Handling

Authentication errors are raised as exceptions:

```python
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}")
```

### Security Best Practices

1. **Never commit secrets**: Use environment variables for keys/secrets
2. **Use HTTPS/TLS**: Always encrypt traffic in production
3. **Rotate keys regularly**: Implement key rotation policies
4. **Short token expiration**: Balance security and UX (1-24 hours)
5. **Log auth attempts**: Track successful and failed authentication
6. **Validate all inputs**: Never trust client-provided data
7. **Use strong secrets**: Generate with `secrets.token_urlsafe(32)`
8. **Implement rate limiting**: Prevent brute force attacks

### OAuth 2.0 Authentication

NextMCP includes a complete OAuth 2.0 implementation with PKCE support, ready-to-use providers for GitHub and Google, and a flexible session management system.

#### OAuth Features

- **OAuth 2.0 with PKCE**: Full Authorization Code Flow with PKCE for secure authentication
- **Ready-to-use Providers**: GitHub and Google OAuth providers with sensible defaults
- **Session Management**: Persistent token storage with memory and file-based backends
- **Auth Metadata Protocol**: Servers can announce authentication requirements to MCP hosts
- **Request Enforcement**: Runtime middleware validates tokens, scopes, and permissions
- **Automatic Token Refresh**: Built-in token refresh handling
- **Production-Ready**: Comprehensive testing and error handling

#### Quick Start with Google OAuth

```python
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

```python
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)
```

#### Multi-Provider Support

Use multiple OAuth providers in the same application:

```python
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}
```

#### Session Management

Session stores persist authentication state and handle token refresh:

```python
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 Scopes and Permissions

Validate OAuth scopes at the tool level:

```python
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 Metadata Protocol

The auth metadata protocol allows MCP servers to announce their authentication requirements to host applications (like Claude Desktop, Cursor, etc.):

```python
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)
```

Hosts can query this metadata to:
- Determine if authentication is required
- Identify the OAuth provider and flow type
- Obtain authorization and token URLs
- Request appropriate scopes
- Guide users through authentication

#### Custom OAuth Providers

Create custom OAuth providers for other services:

```python
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()
```

#### Testing OAuth Flows

Use the provided helper to test OAuth flows interactively:

```bash
# 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
```

See the comprehensive OAuth documentation:
- **[ARCHITECTURE.md](docs/ARCHITECTURE.md)** - System design and data flow
- **[OAUTH_TESTING_SETUP.md](docs/OAUTH_TESTING_SETUP.md)** - Setting up OAuth credentials
- **[MIGRATION_GUIDE.md](docs/MIGRATION_GUIDE.md)** - Adding auth to existing servers
- **[HOST_INTEGRATION.md](docs/HOST_INTEGRATION.md)** - Guide for MCP host developers

### Examples

Check out complete authentication examples:

- **`examples/auth/complete_oauth_server.py`** - Production-ready Google OAuth with session management
- **`examples/auth/github_oauth_server.py`** - GitHub OAuth authentication example
- **`examples/auth/google_oauth_server.py`** - Google OAuth authentication example
- **`examples/auth/multi_provider_server.py`** - Multiple OAuth providers in one server
- **`examples/auth/session_management_example.py`** - Advanced session management workflows
- **`examples/auth/combined_auth_server.py`** - OAuth combined with RBAC and permissions
- **`examples/auth/manifest_server.py`** - Permission manifests with OAuth
- **`examples/auth/oauth_token_helper.py`** - Interactive OAuth testing tool
- **`examples/auth_api_key/`** - API key authentication with role-based access
- **`examples/auth_jwt/`** - JWT token authentication with login endpoint
- **`examples/auth_rbac/`** - Advanced RBAC with fine-grained permissions

## Core Concepts

### Creating an Application

```python
from nextmcp import NextMCP

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

### Registering Tools

```python
@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}
```

### Adding Middleware

Middleware wraps your tools to add cross-cutting functionality.

#### Global Middleware (applied to all tools)

```python
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
```

#### Tool-specific Middleware

```python
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}"
```

### Built-in Middleware

NextMCP includes several production-ready middleware:

- **`log_calls`** - Log all tool invocations with timing
- **`error_handler`** - Catch exceptions and return structured errors
- **`require_auth(valid_keys)`** - API key authentication
- **`rate_limit(max_calls, time_window)`** - Rate limiting
- **`cache_results(ttl_seconds)`** - Response caching
- **`validate_inputs(**validators)`** - Custom input validation
- **`timeout(seconds)`** - Execution timeout

All middleware also have async variants (e.g., `log_calls_async`, `error_handler_async`, etc.) for use with async tools.

### Async Support

NextMCP has full support for async/await patterns, allowing you to build high-performance tools that can handle concurrent I/O operations.

#### Basic Async Tool

```python
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"}
```

#### Async Middleware

Use async middleware variants for async tools:

```python
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}
```

#### Concurrent Operations

The real power of async is handling multiple operations concurrently:

```python
@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}
```

**Performance Comparison:**
- Sequential: 4 sources × 0.1s = 0.4s
- Concurrent (async): ~0.1s (all at once!)

#### Mixed Sync and Async Tools

You can have both sync and async tools in the same application:

```python
@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
```

#### When to Use Async

**Use async for:**
- HTTP API calls (with `httpx`, `aiohttp`)
- Database queries (with `asyncpg`, `motor`)
- File I/O operations
- Multiple concurrent operations
- WebSocket connections

**Stick with sync for:**
- CPU-bound operations (heavy computations)
- Simple operations with no I/O
- When third-party libraries don't support async

See `examples/async_weather_bot/` for a complete async example.

### Schema Validation with Pydantic

```python
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}
```

### Prompts

Prompts are user-driven workflow templates that guide AI interactions. They're explicitly invoked by users (not automatically by the AI) and can reference available tools and resources.

#### Basic Prompts

```python
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
    """
```

#### Prompts with Argument Completion

```python
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()]
```

#### Async Prompts

```python
@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}"
```

**When to use prompts:**
- Guide complex multi-step workflows
- Provide templates for common tasks
- Structure AI interactions
- Reference available tools and resources

See `examples/knowledge_base/` for a complete example using prompts.

### Resources

Resources provide read-only access to contextual data through unique URIs. They're application-driven and give the AI access to information without triggering actions.

#### Direct Resources

```python
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
    }
```

#### Resource Templates

Templates allow parameterized access to dynamic resources:

```python
@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"]
```

#### Subscribable Resources

Resources can notify subscribers when they change:

```python
@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")
```

#### Async Resources

```python
@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")
```

**When to use resources:**
- Provide read-only data access
- Expose configuration and settings
- Share application state
- Offer real-time data feeds (with subscriptions)

**Resource URIs can use any scheme:**
- `file://` - File system access
- `config://` - Configuration data
- `db://` - Database queries
- `api://` - External API data
- Custom schemes for your use case

See `examples/knowledge_base/` for a complete example using resources and templates.

### Configuration

NextMCP supports multiple configuration sources with automatic merging:

```python
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**:
```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 Transport

NextMCP supports WebSocket transport for real-time, bidirectional communication - perfect for chat applications, live updates, and interactive tools.

#### Server Setup

```python
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)
```

##