Metadata-Version: 2.5
Name: http-mcp
Version: 0.18.0
Summary: This is a HTTP implementation of the MCP protocol
Project-URL: Homepage, https://github.com/yeison-liscano/http_mcp
Project-URL: Issues, https://github.com/yeison-liscano/http_mcp/issues
Author: Yeison Liscano
Maintainer: Yeison Liscano
License-Expression: MIT
License-File: LICENSE
Keywords: automation,http,llm,mcp
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.12
Requires-Dist: pydantic<3.0.0,>=2.13.4
Requires-Dist: starlette<2.0.0,>=1.6.0
Requires-Dist: uvicorn<0.53.0,>=0.52.4
Provides-Extra: auth
Description-Content-Type: text/markdown

# Simple HTTP MCP Server Implementation

This project provides a lightweight server implementation for the Model Context
Protocol (MCP) over HTTP. It allows you to expose Python functions as tools and
prompts that can be discovered and executed remotely via a JSON-RPC interface.
It is intended to be used with a Starlette or FastAPI application (see
[demo](https://github.com/yeison-liscano/demo_http_mcp)).

## Table of Contents

- [Features](#features)
- [Installation](#installation)
- [Server Architecture](#server-architecture)
- [Protocol Version](#protocol-version)
  - [The 2026-07-28 Request Shape](#the-2026-07-28-request-shape)
  - [Caching Hints](#caching-hints)
  - [Origin Validation](#origin-validation)
  - [Mirroring Tool Parameters into Headers](#mirroring-tool-parameters-into-headers)
- [Tools](#tools)
  - [Basic Tool Example](#basic-tool-example)
  - [Tools Without Arguments](#tools-without-arguments)
  - [Tools with Error Handling](#tools-with-error-handling)
  - [Tools with Authorization Scopes](#tools-with-authorization-scopes)
- [Server State Management](#server-state-management)
- [Request Access](#request-access)
- [Prompts](#prompts)
  - [Basic Prompt Example](#basic-prompt-example)
  - [Prompts Without Arguments](#prompts-without-arguments)
  - [Prompts with Lifespan State](#prompts-with-lifespan-state)
  - [Prompts with Authorization Scopes](#prompts-with-authorization-scopes)
- [STDIO Transport](#stdio-transport)
- [Authentication and Authorization](#authentication-and-authorization)
- [OAuth 2.1 Authorization (auth_mcp)](#oauth-21-authorization-auth_mcp)
- [API Reference](#api-reference)
- [Security Surfaces by Endpoint](#security-surfaces-by-endpoint)
- [License](#license)

## Features

- **MCP Protocol Compliant**: Implements the MCP specification for tool and
  prompts discovery and execution. No support for notifications.
- **One Protocol Revision**: Speaks the stateless `2026-07-28` revision only —
  `server/discover`, per-request `_meta`, no handshake, no session. A single
  dispatch path means a request cannot select weaker handling by declaring an
  older revision.
- **HTTP and STDIO Transport**: Uses HTTP (POST requests) or STDIO for
  communication.
- **Async Support**: Built on `Starlette` or `FastAPI` for asynchronous request
  handling.
- **Type-Safe**: Leverages `Pydantic` for robust data validation and
  serialization.
- **Server State Management**: Access shared state through the lifespan context
  using the `get_state_key` method.
- **Request Access**: Access the incoming request object from your tools and
  prompts.
- **Authorization Scopes**: Support for scope-based authorization using
  Starlette's authentication system.
- **Error Handling**: Tools can optionally return error messages instead of
  raising exceptions.
- **OAuth 2.1 Authorization**: Optional `auth_mcp` package with Bearer token
  validation, Protected Resource Metadata (RFC 9728), and `WWW-Authenticate`
  error responses. Install with `pip install http-mcp[auth]`.

## Server Architecture

The library provides a single `MCPServer` class that uses lifespan to manage
shared state across the entire application lifecycle.

### MCPServer

The `MCPServer` is designed to work with Starlette's lifespan system for
managing shared server state.

**Key Characteristics:**

- **Lifespan Based**: Uses Starlette's lifespan events to initialize and manage
  shared server state
- **Application-Level State**: State persists across the entire application
  lifecycle, not per-request
- **Flexible**: Can be used with any custom context class stored in the lifespan
  state

**Constructor Parameters:**

- `name` (str): The name of your MCP server
- `version` (str): The version of your MCP server
- `tools` (tuple[Tool, ...]): Tuple of tools to expose (default: empty tuple)
- `prompts` (tuple[Prompt, ...]): Tuple of prompts to expose (default: empty
  tuple)
- `instructions` (str | None): Optional instructions for AI assistants on how to
  use this server
- `cache_ttl_ms` (int): Freshness hint in milliseconds sent with `tools/list`,
  `prompts/list`, and `server/discover` results (default: `300000`). Use `0` to
  tell clients never to cache. See [Caching Hints](#caching-hints).
- `cache_scope` ("public" | "private" | None): Whether shared caches may reuse
  those results across authorization contexts. Derived automatically when
  omitted. See [Caching Hints](#caching-hints).
- `allowed_origins` (tuple[str, ...]): Origins the HTTP transport accepts
  (default: empty, meaning the check is disabled). See
  [Origin Validation](#origin-validation).
- `require_origin` (bool): Whether a request carrying no `Origin` header at all
  is refused when `allowed_origins` is set (default: `False`). See
  [Origin Validation](#origin-validation).

**Example Usage:**

```python
import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
from dataclasses import dataclass, field
from starlette.applications import Starlette
from http_mcp.server import MCPServer

@dataclass
class Context:
    call_count: int = 0
    user_preferences: dict = field(default_factory=dict)

class State(TypedDict):
    context: Context

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    yield {"context": Context()}

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    prompts=my_prompts,
    instructions="Optional instructions for AI assistants on how to use this server"
)

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)
```

## Protocol Version

The server implements exactly one protocol revision, `2026-07-28`, and every
request travels the same path. There is no version negotiation and no second set
of rules a request can select into.

> **Breaking change in 0.17.0.** Support for the session-based revisions
> `2025-11-25`, `2025-06-18`, and `2025-03-26` was removed, along with
> `initialize`, `notifications/initialized`, and `ping`. A client that speaks
> only those revisions can no longer talk to this server. Serving one revision
> is also what makes the request-metadata headers below trustworthy: while two
> eras coexisted, a request could skip the header checks by declaring the older
> one, so an intermediary routing on `Mcp-Method` could be desynchronised from
> the server acting on the body.

> **Breaking change in 0.18.0.** `ServerInterface.get_tool_input_schema` now
> takes the `Request`, so authorization scopes are honoured before dispatch;
> implementations of the interface must update, while `MCPServer` users are
> unaffected. Mirrored `Mcp-Param-*` values are compared textually rather than
> numerically, so a header reading `3.0` for `"replicas": 3` now gets `-32020`.
> Every `notifications/*` method returns 404, the revision defining none.

### The 2026-07-28 Request Shape

The revision has no session concept. In practice:

- **No handshake.** Every request restates its protocol version and the client's
  capabilities in `_meta`:

  ```json
  {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "get_weather",
      "arguments": { "location": "Seattle, WA" },
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": { "name": "ExampleClient", "version": "1.0.0" },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }
  ```

  `protocolVersion` and `clientCapabilities` are required; omitting either gets
  a `-32602` and HTTP 400. Any other version gets a `-32022` whose
  `data.supported` lists the one revision this server speaks.

- **`server/discover` replaces `initialize` for capability discovery.** It
  reports the supported version, capabilities, instructions, and server identity
  in one call, and is answered without any prior request:

  ```json
  {
    "resultType": "complete",
    "supportedVersions": ["2026-07-28"],
    "capabilities": { "tools": { "listChanged": false }, "prompts": { "listChanged": false } },
    "_meta": { "io.modelcontextprotocol/serverInfo": { "name": "my-server", "version": "1.0.0" } },
    "ttlMs": 300000,
    "cacheScope": "public"
  }
  ```

- **Every result carries `resultType: "complete"`** and a `_meta` block naming
  the server.

- **`initialize`, `notifications/initialized`, `ping`, and `logging/setLevel` do
  not exist**, along with the session and SSE resumption machinery. They return
  `-32601` with HTTP 404. JSON-RPC notifications — a `notifications/*` message
  with no `id` — still get `202 Accepted` and no body, because JSON-RPC forbids
  responding to them.

- **Required request headers.** Every POST must send `MCP-Protocol-Version` and
  `Mcp-Method`, plus `Mcp-Name` on `tools/call` and `prompts/get`. Each must
  match the corresponding body value, or the request is rejected with `-32020`
  (`HeaderMismatch`) and HTTP 400 — this stops a proxy routing on one value
  while the server acts on another. Values that cannot be expressed as plain
  ASCII use the `=?base64?...?=` envelope, which the server decodes before
  comparing.

- **Unknown tools and prompts report `-32602`**, which is what the tools and
  prompts specs prescribe. `-32002` was retired by this revision.

- **`Mcp-Session-Id` and `Last-Event-ID` are ignored**, and `GET`/`DELETE` on
  the MCP endpoint return `405 Method Not Allowed`.

Multi round-trip requests (elicitation, sampling, roots) and
`subscriptions/listen` are not implemented: this server exposes no
client-input-dependent features and declares `listChanged: false`, so neither
applies to it.

### Caching Hints

`tools/list`, `prompts/list`, and `server/discover` results carry `ttlMs` and
`cacheScope` so clients can avoid re-fetching a list that has not changed:

```python
mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    cache_ttl_ms=300_000,   # clients may treat the list as fresh for 5 minutes
    cache_scope="public",   # shared caches may serve it to any caller
)
```

Tools and prompts are fixed when `MCPServer` is constructed, so `ttlMs` really
bounds how long a client may miss a redeploy rather than how long the data is
stable. Set it to `0` to ask clients never to cache.

`cache_scope` is derived when you omit it: `"private"` if any tool or prompt is
scope-restricted — the list then varies per caller, so a shared cache must not
reuse it across authorization contexts — and `"public"` otherwise. Override it
if your deployment knows better. Note that `cacheScope` governs caching only; it
is never a substitute for the per-tool scope checks.

### Origin Validation

Browsers attach an `Origin` header, which is what lets a server refuse requests
smuggled in by DNS rebinding. The check is off by default so existing
deployments keep working; turn it on wherever the endpoint is reachable from a
browser:

```python
mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    allowed_origins=("https://app.example.com",),
)
```

A request whose `Origin` is present and not on the list gets `403 Forbidden`.
Requests with no `Origin` at all — ordinary non-browser clients — are unaffected
by default, because browsers always send `Origin` on a POST and the rebinding
threat model does not cover clients that are not browsers.

If the endpoint should only ever serve browser traffic, add `require_origin` to
refuse a request that omits the header too, which makes the allowlist mandatory
rather than advisory:

```python
mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    allowed_origins=("https://app.example.com",),
    require_origin=True,
)
```

`require_origin` does nothing on its own — it only tightens an allowlist that is
already configured. When running locally, also bind to `127.0.0.1` rather than
`0.0.0.0`.

### Mirroring Tool Parameters into Headers

A tool can ask clients to copy specific argument values into `Mcp-Param-*`
headers, so proxies can route or rate-limit on them without parsing the body.
Annotate the field with `x-mcp-header`:

```python
from pydantic import BaseModel, Field

class ExecuteSQLInput(BaseModel):
    region: str = Field(
        description="The region to execute the query in",
        json_schema_extra={"x-mcp-header": "Region"},
    )
    query: str = Field(description="The SQL query to execute")
```

A conforming client then sends `Mcp-Param-Region: us-west1` alongside the call,
and the server verifies it against the body — rejecting the request with
`-32020` if the header is missing, contradicts the argument, or is sent when the
argument is absent. `Mcp-Param-*` headers that no annotation claims are ignored,
as intermediaries are expected to forward unrecognised ones untouched.

The comparison is textual, against the value as JSON writes it: for
`"replicas": 3` the header must read exactly `3`, not `3.0`, `+3`, or `3`.
Numeric coercion would call those equal while an intermediary routing on the raw
header string saw something else, which is the desync the mirroring exists to
prevent.

Only `string`, `integer`, and `boolean` fields reachable through a plain chain
of object properties can be annotated, and no two fields may claim the same
header name — a collision is refused when the server is constructed, because
keeping one of the two annotations would leave the other silently unenforced. Do
not annotate sensitive values: header contents are visible to every intermediary
on the path.

## Tools

Tools are the functions that can be called by the client.

### Basic Tool Example

1. **Define the arguments and output for the tools:**

```python
# app/tools/models.py
from pydantic import BaseModel, Field

class GreetInput(BaseModel):
    question: str = Field(description="The question to answer")

class GreetOutput(BaseModel):
    answer: str = Field(description="The answer to the question")

# Note: the description on Field will be passed when listing the tools.
# Having a description is optional, but it's recommended to provide one.
```

2. **Define the tools:**

```python
# app/tools/tools.py
from http_mcp.types import Arguments

from app.tools.models import GreetInput, GreetOutput

def greet(args: Arguments[GreetInput]) -> GreetOutput:
    return GreetOutput(answer=f"Hello, {args.inputs.question}!")

```

```python
# app/tools/__init__.py

from http_mcp.types import Tool
from app.tools.models import GreetInput, GreetOutput
from app.tools.tools import greet

TOOLS = (
    Tool(
        func=greet,
        inputs=GreetInput,
        output=GreetOutput,
    ),
)

__all__ = ["TOOLS"]

```

3. **Instantiate the server:**

```python
# app/main.py
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.tools import TOOLS

mcp_server = MCPServer(tools=TOOLS, name="test", version="1.0.0")

app = Starlette()
app.mount(
    "/mcp",
    mcp_server.app,
)
```

### Tools Without Arguments

You can define tools that don't require any input arguments:

```python
from datetime import UTC, datetime
from pydantic import BaseModel, Field
from http_mcp.types import Tool

class GetTimeOutput(BaseModel):
    time: str = Field(description="The current time")

async def get_time() -> GetTimeOutput:
    """Get the current time."""
    return GetTimeOutput(time=datetime.now(UTC).strftime("%H:%M:%S"))

TOOLS = (
    Tool(
        func=get_time,
        inputs=type(None),  # No arguments required
        output=GetTimeOutput,
    ),
)
```

Alternatively, you can use the `NoArguments` class for better clarity:

```python
from http_mcp.types import Arguments, NoArguments, Tool

class SimpleOutput(BaseModel):
    success: bool = Field(description="Whether the operation was successful")

def simple_tool(args: Arguments[NoArguments]) -> SimpleOutput:
    """A simple tool with no arguments."""
    # You can still access request and state
    context = args.get_state_key("context", Context)
    return SimpleOutput(success=True)

TOOLS = (
    Tool(
        func=simple_tool,
        inputs=NoArguments,
        output=SimpleOutput,
    ),
)
```

### Tools with Error Handling

Tools can optionally return error messages instead of raising exceptions:

```python
from pydantic import BaseModel, Field
from http_mcp.types import Arguments, Tool
from http_mcp.exceptions import ToolInvocationError

class RiskyToolInput(BaseModel):
    value: int = Field(description="An integer value")

class RiskyToolOutput(BaseModel):
    result: str = Field(description="The result of the operation")

def risky_tool(args: Arguments[RiskyToolInput]) -> RiskyToolOutput:
    """A tool that might fail."""
    if args.inputs.value < 0:
        raise ToolInvocationError("risky_tool", "Value must be positive")
    return RiskyToolOutput(result=f"Success: {args.inputs.value}")

TOOLS = (
    Tool(
        func=risky_tool,
        inputs=RiskyToolInput,
        output=RiskyToolOutput,
        return_error_message=True,  # Return ErrorMessage instead of raising
    ),
)
```

When `return_error_message=True`, the tool will return an `ErrorMessage` model
with the error details instead of raising a `ToolInvocationError`.

### Tools with Authorization Scopes

You can restrict tool access based on authentication scopes:

```python
from http_mcp.exceptions import ToolInvocationError
from http_mcp.types import Arguments, NoArguments, Tool
from starlette.authentication import has_required_scope

class SecureOutput(BaseModel):
    message: str = Field(description="A secure message")

def private_tool(args: Arguments[NoArguments]) -> SecureOutput:
    """A tool that requires authentication."""
    if not has_required_scope(args.request, ("private",)):
        raise ToolInvocationError("private_tool", "Insufficient scope")
    return SecureOutput(message="This is private data")

def admin_tool(args: Arguments[NoArguments]) -> SecureOutput:
    """A tool that requires admin or superuser scope."""
    if not has_required_scope(args.request, ("admin", "superuser")):
        raise ToolInvocationError("admin_tool", "Insufficient scope")
    return SecureOutput(message="This is admin data")

TOOLS = (
    Tool(
        func=private_tool,
        inputs=NoArguments,
        output=SecureOutput,
        scopes=("private",),  # Only accessible with 'private' scope
    ),
    Tool(
        func=admin_tool,
        inputs=NoArguments,
        output=SecureOutput,
        scopes=("admin", "superuser"),  # Accessible with either scope
    ),
)
```

Note: You need to set up authentication middleware in your Starlette app for
scopes to work properly. The `scopes` field on `Tool` is the primary
authorization gate — the framework filters tools by scope before invocation. The
`raise ToolInvocationError(...)` calls inside the tool functions above are
optional defense-in-depth checks that return a proper error response to the
client instead of silently failing.

## Server State Management

The server uses Starlette's lifespan system to manage shared state across the
entire application lifecycle. State is initialized when the application starts
and persists until it shuts down. Context is accessed through the
`get_state_key` method on the `Arguments` object.

This is useful for sharing resources like database connection pools, HTTP
clients, caches, or any application state across tools.

### Database Connection Pool

The most common pattern — initialize a connection pool at startup, share it
across all tools, and close it on shutdown:

```python
# app/context.py
from dataclasses import dataclass
import asyncpg

@dataclass
class AppContext:
    db: asyncpg.Pool
```

```python
# app/main.py
import contextlib
import os
from collections.abc import AsyncIterator
from typing import TypedDict
import asyncpg
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.context import AppContext

class State(TypedDict):
    ctx: AppContext

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    pool = await asyncpg.create_pool(os.environ["DATABASE_URL"])
    yield {"ctx": AppContext(db=pool)}
    await pool.close()

mcp_server = MCPServer(tools=TOOLS, name="my-server", version="1.0.0")

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)
```

```python
# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class GetUserInput(BaseModel):
    user_id: int = Field(description="The user ID to look up")

class GetUserOutput(BaseModel):
    name: str = Field(description="The user's name")
    email: str = Field(description="The user's email")

async def get_user(args: Arguments[GetUserInput]) -> GetUserOutput:
    """Look up a user by ID."""
    ctx = args.get_state_key("ctx", AppContext)
    row = await ctx.db.fetchrow(
        "SELECT name, email FROM users WHERE id = $1",
        args.inputs.user_id,
    )
    return GetUserOutput(name=row["name"], email=row["email"])
```

### Shared HTTP Client

Share a single `httpx.AsyncClient` across tools to reuse connections and
configure base URLs, headers, or timeouts once:

```python
# app/context.py
from dataclasses import dataclass
import httpx

@dataclass
class AppContext:
    http_client: httpx.AsyncClient
```

```python
# app/main.py
import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
import httpx
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.context import AppContext

class State(TypedDict):
    ctx: AppContext

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": "Bearer <token>"},
    ) as client:
        yield {"ctx": AppContext(http_client=client)}

mcp_server = MCPServer(tools=TOOLS, name="my-server", version="1.0.0")

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)
```

```python
# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class SearchInput(BaseModel):
    query: str = Field(description="The search query")

class SearchOutput(BaseModel):
    results: list[str] = Field(description="Search result titles")

async def search(args: Arguments[SearchInput]) -> SearchOutput:
    """Search via an external API."""
    ctx = args.get_state_key("ctx", AppContext)
    resp = await ctx.http_client.get("/search", params={"q": args.inputs.query})
    resp.raise_for_status()
    return SearchOutput(results=[r["title"] for r in resp.json()["items"]])
```

### In-Memory Cache

Share mutable state like caches or counters across tool invocations within the
same server lifecycle:

```python
# app/context.py
from dataclasses import dataclass, field

@dataclass
class AppContext:
    cache: dict[str, str] = field(default_factory=dict)
    request_count: int = 0
```

```python
# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class LookupInput(BaseModel):
    key: str = Field(description="The cache key to look up")

class LookupOutput(BaseModel):
    value: str | None = Field(description="The cached value, or null if not found")
    total_requests: int = Field(description="Total requests served")

async def lookup(args: Arguments[LookupInput]) -> LookupOutput:
    """Look up a value in the cache."""
    ctx = args.get_state_key("ctx", AppContext)
    ctx.request_count += 1
    return LookupOutput(
        value=ctx.cache.get(args.inputs.key),
        total_requests=ctx.request_count,
    )
```

All tools sharing the same `AppContext` instance see each other's writes
immediately, since the lifespan yields a single shared object.

Note: Plain `dict` and `int` are not thread-safe. If your tools run concurrently
(e.g., sync tools dispatched via threads), protect shared mutable state with an
`asyncio.Lock` or use thread-safe data structures.

## Request Access

You can access the incoming request object from your tools. The request object
is passed to each tool call and can be used to access headers, cookies, and
other request data (e.g. request.state, request.scope).

```python
from pydantic import BaseModel, Field
from http_mcp.types import Arguments

class MyToolArguments(BaseModel):
    question: str = Field(description="The question to answer")

class MyToolOutput(BaseModel):
    answer: str = Field(description="The answer to the question")


async def my_tool(args: Arguments[MyToolArguments]) -> MyToolOutput:
    # Access the request
    auth_header = args.request.headers.get("Authorization")
    ...

    return MyToolOutput(answer=f"Hello, {args.inputs.question}!")

# Use MCPServer:
from http_mcp.server import MCPServer

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=(my_tool,),
)
```

## Prompts

You can add interactive templates that are invoked by user choice. Prompts now
support lifespan state access, similar to tools.

### Basic Prompt Example

1. **Define the arguments for the prompts:**

```python
from pydantic import BaseModel, Field

from http_mcp.types import Arguments, Prompt, PromptMessage, TextContent


class GetAdvice(BaseModel):
    topic: str = Field(description="The topic to get advice on")
    include_actionable_steps: bool = Field(
        description="Whether to include actionable steps in the advice", default=False
    )


def get_advice(args: Arguments[GetAdvice]) -> tuple[PromptMessage, ...]:
    """Get advice on a topic."""
    template = """
    You are a helpful assistant that can give advice on {topic}.
    """
    if args.inputs.include_actionable_steps:
        template += """
        The advice should include actionable steps.
        """
    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text=template.format(topic=args.inputs.topic)
            ),
        ),
    )


PROMPTS = (
    Prompt(
        func=get_advice,
        arguments_type=GetAdvice,
    ),
)
```

2. **Instantiate the server:**

```python
from starlette.applications import Starlette

from app.prompts import PROMPTS
from http_mcp.server import MCPServer

app = Starlette()
mcp_server = MCPServer(tools=(), prompts=PROMPTS, name="test", version="1.0.0")

app.mount(
    "/mcp",
    mcp_server.app,
)
```

### Prompts Without Arguments

You can define prompts that don't require any input arguments:

```python
from http_mcp.types import Prompt, PromptMessage, TextContent

def help_prompt() -> tuple[PromptMessage, ...]:
    """Use this prompt to get general help."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text="You are a helpful assistant. Help the user with their task."
            ),
        ),
    )

PROMPTS = (
    Prompt(
        func=help_prompt,
        arguments_type=type(None),  # No arguments required
    ),
)
```

Alternatively, you can use the `NoArguments` class:

```python
from http_mcp.types import Arguments, NoArguments, Prompt, PromptMessage, TextContent

def help_prompt_with_context(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Use this prompt to get help with access to context."""
    # You can still access request and state
    context = args.get_state_key("context", Context)
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="You are a helpful assistant."),
        ),
    )

PROMPTS = (
    Prompt(
        func=help_prompt_with_context,
        arguments_type=NoArguments,
    ),
)
```

### Prompts with Lifespan State

```python
from pydantic import BaseModel, Field
from http_mcp.types import Arguments, Prompt, PromptMessage, TextContent
from app.context import Context

class GetAdvice(BaseModel):
    topic: str = Field(description="The topic to get advice on")

def get_advice_with_context(args: Arguments[GetAdvice]) -> tuple[PromptMessage, ...]:
    """Get advice on a topic with context awareness."""
    # Access the context from lifespan state
    context = args.get_state_key("context", Context)
    called_tools = context.get_called_tools()
    template = """
    You are a helpful assistant that can give advice on {topic}.
    Previously called tools: {tools}
    """

    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text=template.format(
                    topic=args.inputs.topic,
                    tools=", ".join(called_tools) if called_tools else "none"
                )
            )
        ),
    )

PROMPTS_WITH_CONTEXT = (
    Prompt(
        func=get_advice_with_context,
        arguments_type=GetAdvice,
    ),
)
```

### Prompts with Authorization Scopes

You can restrict prompt access based on authentication scopes:

```python
from http_mcp.types import Arguments, NoArguments, Prompt, PromptMessage, TextContent

def private_prompt(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Private prompt that is only accessible to authenticated users."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="This is a private prompt."),
        ),
    )

def admin_prompt(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Admin prompt accessible to users with admin or superuser scope."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="This is an admin prompt."),
        ),
    )

PROMPTS = (
    Prompt(
        func=private_prompt,
        arguments_type=NoArguments,
        scopes=("private",),  # Only accessible with 'private' scope
    ),
    Prompt(
        func=admin_prompt,
        arguments_type=NoArguments,
        scopes=("admin", "superuser"),  # Accessible with either scope
    ),
)
```

Note: You need to set up authentication middleware in your Starlette app for
scopes to work properly.

## STDIO Transport

In addition to HTTP transport, the server supports STDIO transport for
communication. This is useful for command-line applications and integrations
that communicate through standard input/output.

### Using STDIO Transport

```python
import asyncio
import os
from http_mcp.server import MCPServer
from app.tools import TOOLS
from app.prompts import PROMPTS

mcp_server = MCPServer(
    tools=TOOLS,
    prompts=PROMPTS,
    name="test",
    version="1.0.0"
)

# Run the server with STDIO transport
async def main() -> None:
    request_headers = {
        "Authorization": f"Bearer {os.getenv('MCP_TOKEN', '')}",
        "X-Custom-Header": "value",
    }
    await mcp_server.serve_stdio(request_headers)

asyncio.run(main())
```

The `request_headers` parameter allows you to pass headers that will be included
in the request context, enabling authentication and other header-based features
even when using STDIO transport.

## Authentication and Authorization

The library integrates with Starlette's authentication system to provide
scope-based authorization for tools and prompts.

### Setting Up Authentication Middleware

```python
import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials,
    AuthenticationBackend,
    BaseUser,
    SimpleUser,
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import HTTPConnection

from http_mcp.server import MCPServer
from app.context import Context
from app.tools import TOOLS
from app.prompts import PROMPTS


class BasicAuthBackend(AuthenticationBackend):
    def __init__(self, granted_scopes: tuple[str, ...] = ("authenticated",)) -> None:
        self.granted_scopes = granted_scopes
        super().__init__()

    async def authenticate(
        self, conn: HTTPConnection
    ) -> tuple[AuthCredentials, BaseUser] | None:
        # Implement your authentication logic here
        # For example, check Bearer token, API key, etc.
        auth_header = conn.headers.get("Authorization")
        if not auth_header:
            return None

        # Validate token and return credentials with scopes
        return AuthCredentials(self.granted_scopes), SimpleUser("username")


class State(TypedDict):
    context: Context


@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    yield {"context": Context()}


mcp_server = MCPServer(
    tools=TOOLS,
    prompts=PROMPTS,
    name="test",
    version="1.0.0"
)

app = Starlette(
    lifespan=lifespan,
    middleware=[
        Middleware(
            AuthenticationMiddleware,
            backend=BasicAuthBackend(granted_scopes=("private", "admin")),
        ),
    ],
)
app.mount("/mcp", mcp_server.app)
```

### How Scopes Work

1. **Authentication Middleware**: The middleware authenticates each request and
   assigns scopes to the user through `AuthCredentials`.

1. **Tool/Prompt Scopes**: When defining tools or prompts, you can specify
   required scopes using the `scopes` parameter.

1. **Access Control**: The server automatically filters tools and prompts based
   on the user's granted scopes. Tools and prompts without the required scopes
   are not visible in listings and cannot be invoked.

1. **Multiple Scopes**: If you specify multiple scopes (e.g.,
   `scopes=("admin", "superuser")`), the user needs at least one of those scopes
   to access the tool or prompt.

## API Reference

### Tool Class

The `Tool` class is used to define tools that can be invoked by clients.

**Parameters:**

- `func`: The function to be invoked. Can be sync or async. The function can
  either:
  - Accept an `Arguments[TInputs]` parameter
  - Accept no parameters
- `inputs`: The Pydantic model class for input validation. Use `type(None)` or
  `NoArguments` for tools without inputs
- `output`: The Pydantic model class for output validation
- `return_error_message` (bool): If `True`, tool errors return `ErrorMessage`
  instead of raising exceptions (default: `False`)
- `scopes` (tuple[str, ...]): Required authentication scopes for accessing this
  tool (default: empty tuple)

**Properties:**

- `name`: The function name (derived from `func.__name__`)
- `title`: A human-readable title (derived from the function name)
- `description`: The function's docstring
- `input_schema`: JSON schema for the input parameters
- `output_schema`: JSON schema for the output

### Prompt Class

The `Prompt` class is used to define prompts that can be invoked by clients.

**Parameters:**

- `func`: The function to be invoked. Can be sync or async. The function can
  either:
  - Accept an `Arguments[TArguments]` parameter
  - Accept no parameters
  - Must return `tuple[PromptMessage, ...]`
- `arguments_type`: The Pydantic model class for argument validation. Use
  `type(None)` or `NoArguments` for prompts without arguments
- `scopes` (tuple[str, ...]): Required authentication scopes for accessing this
  prompt (default: empty tuple)

**Properties:**

- `name`: The function name (derived from `func.__name__`)
- `title`: A human-readable title (derived from the function name)
- `description`: The function's docstring
- `arguments`: Tuple of `PromptArgument` objects defining the prompt's arguments

### Arguments Class

The `Arguments` class is passed to tool and prompt functions to provide access
to inputs, request, and state.

**Parameters:**

- `request`: The Starlette `Request` object
- `inputs`: The validated input/argument data (type depends on the Tool/Prompt
  definition)

**Methods:**

- `get_state_key(key: str, _object_type: type[TKey]) -> TKey`: Access a value
  from the lifespan state. Raises `ServerError` if the key doesn't exist.

### NoArguments Class

An empty Pydantic model that can be used as a clearer alternative to
`type(None)` when defining tools or prompts without arguments.

```python
from http_mcp.types import NoArguments

# Use this instead of type(None)
Tool(func=my_func, inputs=NoArguments, output=MyOutput)
```

## OAuth 2.1 Authorization (auth_mcp)

The `auth_mcp` package adds standards-compliant OAuth 2.1 authorization to your
MCP server. Install with the `auth` extra:

```bash
pip install http-mcp[auth]
```

### Quick Start

```python
from http_mcp.server import MCPServer
from auth_mcp.resource_server import (
    ProtectedMCPAppConfig,
    TokenInfo,
    TokenValidator,
    create_protected_mcp_app,
)
from auth_mcp.types import ProtectedResourceMetadata


class MyTokenValidator(TokenValidator):
    async def validate_token(
        self, token: str, resource: str | None = None
    ) -> TokenInfo | None:
        # Validate against your authorization server
        ...


mcp_server = MCPServer(name="my-server", version="1.0.0", tools=MY_TOOLS)

config = ProtectedMCPAppConfig(
    mcp_server=mcp_server,
    token_validator=MyTokenValidator(),
    resource_endpoint=ProtectedResourceMetadata(
        resource="https://mcp.example.com",
        authorization_servers=("https://auth.example.com",),
    ),
)

app = create_protected_mcp_app(config)
```

This gives you:

- Bearer token validation on all MCP endpoints (secure by default)
- `/.well-known/oauth-protected-resource` discovery endpoint (RFC 9728)
- `WWW-Authenticate` headers on 401/403 with `resource_metadata` parameter
- Security headers (HSTS, nosniff, no-store)
- Optional custom middleware via the `middlewares` parameter

For full documentation, best practices, and security surface details, see
[auth_mcp README](src/auth_mcp/README.md).

## Security Surfaces by Endpoint

### `POST /mcp` — MCP JSON-RPC Endpoint

- **Authentication** — When using `auth_mcp`, Bearer tokens are extracted from
  the `Authorization` header and validated via `TokenValidator`. Tokens
  exceeding 2048 characters or containing characters outside the RFC 6750
  `b64token` pattern are rejected before reaching the validator. Without
  `auth_mcp`, authentication is handled by Starlette's
  `AuthenticationMiddleware`.
- **Authorization** — Scope-based filtering via Starlette's
  `has_required_scope()`. Tools and prompts without matching scopes are hidden
  from listings and blocked on invocation. Request-header validation resolves
  tool schemas through the same scope check, so a caller a tool is hidden from
  cannot learn its `x-mcp-header` arguments from a mismatch message either.
- **Input validation** — JSON-RPC messages validated by Pydantic. Request body
  capped at 4 MB, enforced while reading: an oversized `Content-Length` is
  refused before the body is read at all, and a body that outgrows the cap
  mid-stream stops being buffered at that point. Content-Type strictly checked
  (`application/json` only, media type parameters ignored).
- **Error handling** — Tool and prompt names truncated to 100 characters in
  error messages. Pydantic validation errors sanitized before inclusion in
  responses.
- **Response headers** — `X-Content-Type-Options: nosniff`,
  `Cache-Control: no-store` on all responses. `auth_mcp` additionally adds
  `Strict-Transport-Security: max-age=31536000; includeSubDomains`.

### `GET /.well-known/oauth-protected-resource` — Discovery Endpoint (auth_mcp)

- **Authentication** — Subject to the same auth middleware as `/mcp`. When
  `require_authentication=True` (default), requires a valid token. Set to
  `False` if clients need to discover the authorization server before
  authenticating.
- **Input validation** — Only `GET` allowed; other methods return
  `405 Method Not Allowed`.
- **Output** — Serialized once at startup from a frozen
  `ProtectedResourceMetadata` model. URI fields validated as HTTP/HTTPS URLs via
  Pydantic's `AnyHttpUrl`.

### `WWW-Authenticate` Response Header (auth_mcp)

- **Header injection** — All parameter values (`realm`, `resource_metadata`,
  `scope`, `error`, `error_description`) are sanitized: CR/LF characters
  stripped, backslash and double-quote escaped per RFC 7230 quoted-string rules.
- **Information disclosure** — Error responses use generic messages
  (`"Authentication required"`). The original `AuthenticationError` details are
  discarded. Error codes (`invalid_token` on 401) follow RFC 6750 without
  leaking internal state.

### STDIO Transport

- **Message size** — Capped at 4 MB, matching HTTP transport.
- **Logging** — Messages truncated to 500 characters in debug logs to prevent
  log flooding. Token values are never logged.
- **Headers** — Request headers are converted to proper ASGI
  `list[tuple[bytes, bytes]]` format.

## Installation

Requires **Python 3.12+** (uses PEP 695 type parameter syntax).

Install the package using pip or uv:

```bash
pip install http-mcp
```

With OAuth 2.1 authorization support:

```bash
pip install http-mcp[auth]
```

or

```bash
uv add http-mcp
```

## License

This project is licensed under the MIT License. See the LICENSE file for
details.
