python-style — independently scanned and version-tracked by SaferSkills.
SaferSkills independently audited python-style (Rules) and scored it 100/100 (green). The audit ran 55 deterministic rules across Security, Supply Chain, Maintenance, Transparency, and Community; it found 0 high-severity and 0 lower-severity findings. The full rule-by-rule trace and per-finding evidence are below. Free, methodology-open.
Findings & checks · 0 flagged
Every scanned point with the score it earned and what moved between them.
First recorded scan — no prior version to compare against.
The primary manifest — the file an agent reads to learn what this artifact does.
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).
prompts discovery and execution. No support for notifications.
communication.
Starlette or FastAPI for asynchronous requesthandling.
Pydantic for robust data validation andserialization.
using the get_state_key method.
prompts.
Starlette's authentication system.
raising exceptions.
auth_mcp package with Bearer tokenvalidation, Protected Resource Metadata (RFC 9728), and WWW-Authenticate error responses. Install with pip install http-mcp[auth].
The library provides a single MCPServer class that uses lifespan to manage shared state across the entire application lifecycle.
The MCPServer is designed to work with Starlette's lifespan system for managing shared server state.
Key Characteristics:
shared server state
lifecycle, not per-request
state
Constructor Parameters:
name (str): The name of your MCP serverversion (str): The version of your MCP servertools (tuple[Tool, ...]): Tuple of tools to expose (default: empty tuple)prompts (tuple[Prompt, ...]): Tuple of prompts to expose (default: emptytuple)
instructions (str | None): Optional instructions for AI assistants on how touse 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 are the functions that can be called by the client.
# 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.# 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"]
# 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,
)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 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.
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.
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.
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"])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"]])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.
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,),
)You can add interactive templates that are invoked by user choice. Prompts now support lifespan state access, similar to tools.
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,
),
)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,
)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,
),
)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,
),
)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.
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.
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.
The library integrates with Starlette's authentication system to provide scope-based authorization for tools and prompts.
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)assigns scopes to the user through AuthCredentials.
required scopes using the scopes parameter.
on the user's granted scopes. Tools and prompts without the required scopes are not visible in listings and cannot be invoked.
scopes=("admin", "superuser")), the user needs at least one of those scopes to access the tool or prompt.
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 caneither:
Arguments[TInputs] parameterinputs: The Pydantic model class for input validation. Use type(None) orNoArguments for tools without inputs
output: The Pydantic model class for output validationreturn_error_message (bool): If True, tool errors return ErrorMessageinstead of raising exceptions (default: False)
scopes (tuple[str, ...]): Required authentication scopes for accessing thistool (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 docstringinput_schema: JSON schema for the input parametersoutput_schema: JSON schema for the outputThe 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 caneither:
Arguments[TArguments] parametertuple[PromptMessage, ...]arguments_type: The Pydantic model class for argument validation. Usetype(None) or NoArguments for prompts without arguments
scopes (tuple[str, ...]): Required authentication scopes for accessing thisprompt (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 docstringarguments: Tuple of PromptArgument objects defining the prompt's argumentsThe Arguments class is passed to tool and prompt functions to provide access to inputs, request, and state.
Parameters:
request: The Starlette Request objectinputs: The validated input/argument data (type depends on the Tool/Promptdefinition)
Methods:
get_state_key(key: str, _object_type: type[TKey]) -> TKey: Access a valuefrom the lifespan state. Raises ServerError if the key doesn't exist.
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)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]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:
/.well-known/oauth-protected-resource discovery endpoint (RFC 9728)WWW-Authenticate headers on 401/403 with resource_metadata parametermiddlewares parameterFor full documentation, best practices, and security surface details, see auth_mcp README.
POST /mcp — MCP JSON-RPC Endpointauth_mcp, Bearer tokens are extracted fromthe 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.
has_required_scope(). Tools and prompts without matching scopes are hidden from listings and blocked on invocation.
capped at 4 MB. Content-Type strictly checked (application/json only, media type parameters ignored).
error messages. Pydantic validation errors sanitized before inclusion in responses.
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)/mcp. Whenrequire_authentication=True (default), requires a valid token. Set to False if clients need to discover the authorization server before authenticating.
GET allowed; other methods return405 Method Not Allowed.
ProtectedResourceMetadata model. URI fields validated as HTTP/HTTPS URLs via Pydantic's AnyHttpUrl.
WWW-Authenticate Response Header (auth_mcp)realm, resource_metadata,scope, error, error_description) are sanitized: CR/LF characters stripped, backslash and double-quote escaped per RFC 7230 quoted-string rules.
("Authentication required"). The original AuthenticationError details are discarded. Error codes (invalid_token on 401) follow RFC 6750 without leaking internal state.
log flooding. Token values are never logged.
list[tuple[bytes, bytes]] format.
Requires Python 3.12+ (uses PEP 695 type parameter syntax).
Install the package using pip or uv:
pip install http-mcpWith OAuth 2.1 authorization support:
pip install http-mcp[auth]or
uv add http-mcpThis project is licensed under the MIT License. See the LICENSE file for details.
~30 seconds. Free. No account. Every finding cites a rule and a line of evidence.