# Fastmcp

> |

- **Type:** Skill
- **Install:** `agentstack add skill-kgeminic-claude-skills-1-fastmcp`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [Kgeminic](https://agentstack.voostack.com/s/kgeminic)
- **Installs:** 0
- **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [Kgeminic](https://github.com/Kgeminic)
- **Source:** https://github.com/Kgeminic/claude-skills-1/tree/main/skills/fastmcp

## Install

```sh
agentstack add skill-kgeminic-claude-skills-1-fastmcp
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# FastMCP - Build MCP Servers in Python

FastMCP is a Python framework for building Model Context Protocol (MCP) servers that expose tools, resources, and prompts to Large Language Models like Claude. This skill provides production-tested patterns, error prevention, and deployment strategies for building robust MCP servers.

## Quick Start

### Installation

```bash
pip install fastmcp
# or
uv pip install fastmcp
```

### Minimal Server

```python
from fastmcp import FastMCP

# MUST be at module level for FastMCP Cloud
mcp = FastMCP("My Server")

@mcp.tool()
async def hello(name: str) -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"

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

**Run it:**
```bash
# Local development
python server.py

# With FastMCP CLI
fastmcp dev server.py

# HTTP mode
python server.py --transport http --port 8000
```

## What's New in v2.14.x (December 2025)

### v2.14.2 (December 31, 2024)
- MCP SDK pinned to  dict:
    # Clients see highest version by default
    # Can request specific version
    return {"data": [...]}
```

### Session-Scoped State

```python
@mcp.tool()
async def set_preference(key: str, value: str, ctx: Context) -> dict:
    await ctx.set_state(key, value)  # Persists across session
    return {"saved": True}

@mcp.tool()
async def get_preference(key: str, ctx: Context) -> dict:
    value = await ctx.get_state(key, default=None)
    return {"value": value}
```

### Other Features

- `--reload` flag for auto-restart during development
- Automatic threadpool dispatch for sync functions
- Tool timeouts
- OpenTelemetry tracing
- Component authorization: `@tool(auth=require_scopes("admin"))`

### Migration Guide

**Pin to v2 if not ready**:
```
# requirements.txt
fastmcp dict:  # Use async for I/O
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).json()
```

### Resources
Expose data to LLMs. URI schemes: `data://`, `file://`, `resource://`, `info://`, `api://`, or custom.

```python
@mcp.resource("user://{user_id}/profile")  # Template with parameters
async def get_user(user_id: str) -> dict:  # CRITICAL: param names must match
    return await fetch_user_from_db(user_id)
```

### Prompts
Pre-configured prompts with parameters.

```python
@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
    return f"Analyze {topic} considering: state, challenges, opportunities, recommendations."
```

## Context Features

Inject `Context` parameter (with type hint!) for advanced features:

**Elicitation (User Input):**
```python
from fastmcp import Context

@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
    confirmed = await context.request_elicitation(prompt=f"Confirm {action}?", response_type=str)
    return {"status": "completed" if confirmed.lower() == "yes" else "cancelled"}
```

**Progress Tracking:**
```python
@mcp.tool()
async def batch_import(file_path: str, context: Context) -> dict:
    data = await read_file(file_path)
    for i, item in enumerate(data):
        await context.report_progress(i + 1, len(data), f"Importing {i + 1}/{len(data)}")
        await import_item(item)
    return {"imported": len(data)}
```

**Sampling (LLM calls from tools):**
```python
@mcp.tool()
async def enhance_text(text: str, context: Context) -> str:
    response = await context.request_sampling(
        messages=[{"role": "user", "content": f"Enhance: {text}"}],
        temperature=0.7
    )
    return response["content"]
```

## Background Tasks (v2.14.0+)

Long-running operations that report progress without blocking clients. Uses Docket task scheduler (always enabled in v2.14.0+).

**Basic Usage:**
```python
@mcp.tool(task=True)  # Enable background task mode
async def analyze_large_dataset(dataset_id: str, context: Context) -> dict:
    """Analyze large dataset with progress tracking."""
    data = await fetch_dataset(dataset_id)

    for i, chunk in enumerate(data.chunks):
        # Report progress to client
        await context.report_progress(
            current=i + 1,
            total=len(data.chunks),
            message=f"Processing chunk {i + 1}/{len(data.chunks)}"
        )
        await process_chunk(chunk)

    return {"status": "complete", "records_processed": len(data)}
```

**Task States:** `pending` → `running` → `completed` / `failed` / `cancelled`

**When to Use:**
- Operations taking >30 seconds (LLM timeout risk)
- Batch processing with per-item status updates
- Operations that may need user input mid-execution
- Long-running API calls or data processing

**Known Limitation (v2.14.x)**:
- `statusMessage` from `ctx.report_progress()` is **not forwarded** to clients during background task polling ([GitHub Issue #2904](https://github.com/jlowin/fastmcp/issues/2904))
- Progress messages appear in server logs but not in client UI
- **Workaround**: Use official MCP SDK (`mcp>=1.10.0`) instead of FastMCP for now
- **Status**: Fix pending in [PR #2906](https://github.com/jlowin/fastmcp/pull/2906)

**Important:** Tasks execute through Docket scheduler. Cannot execute tasks through proxies (will raise error).

## Sampling with Tools (v2.14.1+)

Servers can pass tools to `ctx.sample()` for agentic workflows where the LLM can call tools during sampling.

**Agentic Sampling:**
```python
from fastmcp import Context
from fastmcp.sampling import AnthropicSamplingHandler

# Configure sampling handler
mcp = FastMCP("Agent Server")
mcp.add_sampling_handler(AnthropicSamplingHandler(api_key=os.getenv("ANTHROPIC_API_KEY")))

@mcp.tool()
async def research_topic(topic: str, context: Context) -> dict:
    """Research a topic using agentic sampling with tools."""

    # Define tools available during sampling
    research_tools = [
        {
            "name": "search_web",
            "description": "Search the web for information",
            "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}
        },
        {
            "name": "fetch_url",
            "description": "Fetch content from a URL",
            "inputSchema": {"type": "object", "properties": {"url": {"type": "string"}}}
        }
    ]

    # Sample with tools - LLM can call these tools during reasoning
    result = await context.sample(
        messages=[{"role": "user", "content": f"Research: {topic}"}],
        tools=research_tools,
        max_tokens=4096
    )

    return {"research": result.content, "tools_used": result.tool_calls}
```

**Single-Step Sampling:**
```python
@mcp.tool()
async def get_single_response(prompt: str, context: Context) -> dict:
    """Get a single LLM response without tool loop."""

    # sample_step() returns SampleStep for inspection
    step = await context.sample_step(
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )

    return {
        "content": step.content,
        "model": step.model,
        "stop_reason": step.stop_reason
    }
```

**Sampling Handlers:**
- `AnthropicSamplingHandler` - For Claude models (v2.14.1+)
- `OpenAISamplingHandler` - For GPT models

**Known Limitation**:
`ctx.sample()` works when client connects to a single server but fails with "Sampling not supported" error when multiple servers are configured in client. Tools without sampling work fine. ([Community-sourced finding](https://github.com/jlowin/fastmcp/issues/699))

## Storage Backends

Built on `py-key-value-aio` for OAuth tokens, response caching, persistent state.

**Available Backends:**
- **Memory** (default): Ephemeral, fast, dev-only
- **Disk**: Persistent, encrypted with `FernetEncryptionWrapper`, platform-aware (Mac/Windows default)
- **Redis**: Distributed, production, multi-instance
- **Others**: DynamoDB, MongoDB, Elasticsearch, Memcached, RocksDB, Valkey

**Basic Usage:**
```python
from key_value.stores import DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

# Disk (persistent, single instance)
mcp = FastMCP("Server", storage=DiskStore(path="/app/data/storage"))

# Redis (distributed, production)
mcp = FastMCP("Server", storage=RedisStore(
    host=os.getenv("REDIS_HOST"), password=os.getenv("REDIS_PASSWORD")
))

# Encrypted storage (recommended)
mcp = FastMCP("Server", storage=FernetEncryptionWrapper(
    key_value=DiskStore(path="/app/data"),
    fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
))
```

**Platform Defaults:** Mac/Windows use Disk, Linux uses Memory. Override with `storage` parameter.

## Server Lifespans

**⚠️ Breaking Change in v2.13.0**: Lifespan behavior changed from per-session to per-server-instance.

Initialize/cleanup resources once per server (NOT per session) - critical for DB connections, API clients.

```python
from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppContext:
    db: Database
    api_client: httpx.AsyncClient

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Runs ONCE per server instance."""
    db = await Database.connect(os.getenv("DATABASE_URL"))
    api_client = httpx.AsyncClient(base_url=os.getenv("API_BASE_URL"), timeout=30.0)

    try:
        yield AppContext(db=db, api_client=api_client)
    finally:
        await db.disconnect()
        await api_client.aclose()

mcp = FastMCP("Server", lifespan=app_lifespan)

# Access in tools
@mcp.tool()
async def query_db(sql: str, context: Context) -> list:
    app_ctx = context.fastmcp_context.lifespan_context
    return await app_ctx.db.query(sql)
```

**ASGI Integration (FastAPI/Starlette):**
```python
mcp = FastMCP("Server", lifespan=mcp_lifespan)
app = FastAPI(lifespan=mcp.lifespan)  # ✅ MUST pass lifespan!
```

**State Management:**
```python
context.fastmcp_context.set_state(key, value)  # Store
context.fastmcp_context.get_state(key, default=None)  # Retrieve
```

## Middleware System

**8 Built-in Types:** TimingMiddleware, ResponseCachingMiddleware, LoggingMiddleware, RateLimitingMiddleware, ErrorHandlingMiddleware, ToolInjectionMiddleware, PromptToolMiddleware, ResourceToolMiddleware

**Execution Order (order matters!):**
```
Request Flow:
  → ErrorHandlingMiddleware (catches errors)
    → TimingMiddleware (starts timer)
      → LoggingMiddleware (logs request)
        → RateLimitingMiddleware (checks rate limit)
          → ResponseCachingMiddleware (checks cache)
            → Tool/Resource Handler
```

**Basic Usage:**
```python
from fastmcp.middleware import ErrorHandlingMiddleware, TimingMiddleware, LoggingMiddleware

mcp.add_middleware(ErrorHandlingMiddleware())  # First: catch errors
mcp.add_middleware(TimingMiddleware())         # Second: time requests
mcp.add_middleware(LoggingMiddleware(level="INFO"))
mcp.add_middleware(RateLimitingMiddleware(max_requests=100, window_seconds=60))
mcp.add_middleware(ResponseCachingMiddleware(ttl_seconds=300, storage=RedisStore()))
```

**Custom Middleware:**
```python
from fastmcp.middleware import BaseMiddleware

class AccessControlMiddleware(BaseMiddleware):
    async def on_call_tool(self, tool_name, arguments, context):
        user = context.fastmcp_context.get_state("user_id")
        if user not in self.allowed_users:
            raise PermissionError(f"User not authorized")
        return await self.next(tool_name, arguments, context)
```

**Hook Hierarchy:** `on_message` (all) → `on_request`/`on_notification` → `on_call_tool`/`on_read_resource`/`on_get_prompt` → `on_list_*` (list operations)

## Server Composition

**Two Strategies:**

1. **`import_server()`** - Static snapshot: One-time copy at import, changes don't propagate, fast (no runtime delegation). Use for: Finalized component bundles.

2. **`mount()`** - Dynamic link: Live runtime link, changes immediately visible, runtime delegation (slower). Use for: Modular runtime composition.

**Basic Usage:**
```python
# Import (static)
main_server.import_server(api_server)  # One-time copy

# Mount (dynamic)
main_server.mount(api_server, prefix="api")  # Tools: api.fetch_data
main_server.mount(db_server, prefix="db")    # Resources: resource://db/path
```

**Tag Filtering:**
```python
@api_server.tool(tags=["public"])
def public_api(): pass

main_server.import_server(api_server, include_tags=["public"])  # Only public
main_server.mount(api_server, prefix="api", exclude_tags=["admin"])  # No admin
```

**Resource Prefix Formats:**
- **Path** (default since v2.4.0): `resource://prefix/path`
- **Protocol** (legacy): `prefix+resource://path`

```python
main_server.mount(subserver, prefix="api", resource_prefix_format="path")
```

## OAuth & Authentication

**4 Authentication Patterns:**

1. **Token Validation** (`JWTVerifier`): Validate external tokens
2. **External Identity Providers** (`RemoteAuthProvider`): OAuth 2.0/OIDC with DCR
3. **OAuth Proxy** (`OAuthProxy`): Bridge to providers without DCR (GitHub, Google, Azure, AWS, Discord, Facebook)
4. **Full OAuth** (`OAuthProvider`): Complete authorization server

**Pattern 1: Token Validation**
```python
from fastmcp.auth import JWTVerifier

auth = JWTVerifier(issuer="https://auth.example.com", audience="my-server",
                   public_key=os.getenv("JWT_PUBLIC_KEY"))
mcp = FastMCP("Server", auth=auth)
```

**Pattern 3: OAuth Proxy (Production)**
```python
from fastmcp.auth import OAuthProxy
from key_value.stores import RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

auth = OAuthProxy(
    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
    client_storage=FernetEncryptionWrapper(
        key_value=RedisStore(host=os.getenv("REDIS_HOST"), password=os.getenv("REDIS_PASSWORD")),
        fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
    ),
    upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
    upstream_token_endpoint="https://github.com/login/oauth/access_token",
    upstream_client_id=os.getenv("GITHUB_CLIENT_ID"),
    upstream_client_secret=os.getenv("GITHUB_CLIENT_SECRET"),
    enable_consent_screen=True  # CRITICAL: Prevents confused deputy attacks
)
mcp = FastMCP("GitHub Auth", auth=auth)
```

**OAuth Proxy Features:** Token factory pattern (issues own JWTs), consent screens (prevents bypass), PKCE support, RFC 7662 token introspection

**Supported Providers:** GitHub, Google, Azure, AWS Cognito, Discord, Facebook, WorkOS, AuthKit, Descope, Scalekit, OCI (v2.13.1)

**Supabase Provider** (v2.14.2+):
```python
from fastmcp.auth import SupabaseProvider

auth = SupabaseProvider(
    auth_route="/custom-auth",  # Custom auth route (new in v2.14.2)
    # ... other config
)
```

## Icons, API Integration, Cloud Deployment

**Icons:** Add to servers, tools, resources, prompts. Use `Icon(url, size)`, data URIs via `Icon.from_file()` or `Image.to_data_uri()` (v2.13.1).

**API Integration (3 Patterns):**
1. **Manual**: `httpx.AsyncClient` with base_url/headers/timeout
2. **OpenAPI Auto-Gen**: `FastMCP.from_openapi(spec, client, route_maps)` - GET→Resources/Templates, POST/PUT/DELETE→Tools
3. **FastAPI Conversion**: `FastMCP.from_fastapi(app, httpx_client_kwargs)`

**Cloud Deployment Critical Requirements:**
1. ❗ **Module-level server** named `mcp`, `server`, or `app`
2. **PyPI dependencies only** in requirements.txt
3. **Public GitHub repo** (or accessible)
4. **Environment variables** for config

```python
# ✅ CORRECT: Module-level export
mcp = FastMCP("server")  # At module level!

# ❌ WRONG: Function-wrapped
def create_server():
    return FastMCP("server")  # Too late for cloud!
```

**Deployment:** https://fastmcp.cloud → Sign in → Create Project → Select repo → Deploy

**Client Config (Claude Desktop):**
```json
{"mcpServers": {"my-server": {"url": "https://project.fastmcp.app/mcp", "transport": "http"}}}
```

## 30 Common Errors (With Solutions)

### Error 1: Missing Server Object
**Error:** `RuntimeError: No server object found at module level`
**Cause:** Server not exported at module level (FastMCP Cloud requirement)
**Solution:** `mcp = FastM

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [Kgeminic](https://github.com/Kgeminic)
- **Source:** [Kgeminic/claude-skills-1](https://github.com/Kgeminic/claude-skills-1)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** yes
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** yes
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-kgeminic-claude-skills-1-fastmcp
- Seller: https://agentstack.voostack.com/s/kgeminic
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.