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).

Features

MCP Protocol Compliant: Implements the MCP specification for tool and prompts discovery and execution. No support for notifications.
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

Example Usage:

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)

Tools

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

Basic Tool Example

Define the arguments and output for the tools:

# 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.

Define the tools:

# 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}!")

# 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"]

Instantiate the server:

# 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:

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:

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:

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:

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:

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

@dataclass
class AppContext:
    db: asyncpg.Pool

# 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)

# 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:

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

@dataclass
class AppContext:
    http_client: httpx.AsyncClient

# 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)

# 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:

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

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

# 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).

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

Define the arguments for the prompts:

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,
    ),
)

Instantiate the server:

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:

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:

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

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:

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

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

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

Authentication Middleware: The middleware authenticates each request and assigns scopes to the user through AuthCredentials.
Tool/Prompt Scopes: When defining tools or prompts, you can specify required scopes using the scopes parameter.
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.
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.

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:

pip install http-mcp[auth]

Quick Start

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.

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.
Input validation — JSON-RPC messages validated by Pydantic. Request body capped at 4 MB. 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:

pip install http-mcp

With OAuth 2.1 authorization support:

pip install http-mcp[auth]

or

uv add http-mcp

License

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

Name		Name	Last commit message	Last commit date
Latest commit History 77 Commits
.claude/agents		.claude/agents
.cursor/rules		.cursor/rules
.gemini		.gemini
.githooks		.githooks
.github/workflows		.github/workflows
.vscode		.vscode
assets		assets
docs		docs
src		src
tests		tests
.envrc		.envrc
.gitignore		.gitignore
.python-version		.python-version
AGENT.md		AGENT.md
CLAUDE.md		CLAUDE.md
LICENSE		LICENSE
README.md		README.md
fluidattacks_sast_config.yaml		fluidattacks_sast_config.yaml
fluidattacks_sca_config.yaml		fluidattacks_sca_config.yaml
fluidattacks_scanners_config.yaml		fluidattacks_scanners_config.yaml
mypy.ini		mypy.ini
pyproject.toml		pyproject.toml
ruff.toml		ruff.toml
uv.lock		uv.lock

Folders and files

Latest commit

History

Repository files navigation

Simple HTTP MCP Server Implementation

Table of Contents

Features

Server Architecture

MCPServer

Tools

Basic Tool Example

Tools Without Arguments

Tools with Error Handling

Tools with Authorization Scopes

Server State Management

Database Connection Pool

Shared HTTP Client

In-Memory Cache

Request Access

Prompts

Basic Prompt Example

Prompts Without Arguments

Prompts with Lifespan State

Prompts with Authorization Scopes

STDIO Transport

Using STDIO Transport

Authentication and Authorization

Setting Up Authentication Middleware

How Scopes Work

API Reference

Tool Class

Prompt Class

Arguments Class

NoArguments Class

OAuth 2.1 Authorization (auth_mcp)

Quick Start

Security Surfaces by Endpoint

POST /mcp — MCP JSON-RPC Endpoint

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

WWW-Authenticate Response Header (auth_mcp)

STDIO Transport

Installation

License

About

Topics

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases 9

Packages 0

Uh oh!

Uh oh!

Contributors

Uh oh!

Languages

`POST /mcp` — MCP JSON-RPC Endpoint

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

`WWW-Authenticate` Response Header (auth_mcp)

Packages