pip install py-http-mcp-server      # no dependencies are pulled in

from fastapi import FastAPI
from py_http_mcp_server import McpServer

app = FastAPI()
mcp = McpServer(server_name="orders-mcp")

@mcp.tool()
def get_order(order_id: int) -> dict:
    """Returns an order by id."""
    return {"id": order_id, "status": "shipped"}

mcp.mount(app)          # ← serves /mcp alongside your API (like app.MapMcpHttpServer() in .NET)

mcp.mount(flask_app)                                      # Flask: wraps flask_app.wsgi_app

application = mcp.wsgi_middleware(get_wsgi_application()) # Django (wsgi.py)
application = mcp.asgi_middleware(get_asgi_application()) # Django (asgi.py), Quart, ...
app.add_middleware(McpAsgiMiddleware, server=mcp)         # Starlette/FastAPI alternative to mount()

mcp.run(port=8000)                 # stdlib threaded HTTP server
# or: uvicorn main:mcp.asgi_app    /   waitress-serve main:mcp.wsgi_app

# myapp/tools/orders.py
from py_http_mcp_server import mcp_tool

@mcp_tool
def cancel_order(order_id: int) -> str:
    """Cancels an order."""
    ...

# main.py
mcp = McpServer(server_name="orders-mcp").discover("myapp.tools")   # scans the package recursively
mcp.mount(app)

from dataclasses import dataclass
from typing import Annotated
from py_http_mcp_server import McpServer, Param, CancellationToken

server = McpServer(
    server_name="MyMcpServer",
    server_version="1.0.0",
    instructions="Use these tools to query system metrics and data.",
    use_default_cors=True,
    list_cache_ttl_ms=60000,
)

@server.tool("calculate_sum", "Calculates the sum of two numbers.", read_only_hint=True, idempotent_hint=True)
def calculate_sum(a: Annotated[int, Param("First number")], b: Annotated[int, Param("Second number")]) -> int:
    return a + b

@dataclass
class WeatherReport:
    city: str
    temperature_c: float
    condition: str

@server.tool("get_weather", "Returns structured weather data.", use_structured_output=True)
async def get_weather(city: Annotated[str, Param("City", enum=["Seattle", "London"])], ct: CancellationToken) -> WeatherReport:
    return WeatherReport(city, 24.5, "Sunny")

@server.resource("system://status", "System Status", "Current status", "application/json")
def status() -> dict:
    return {"status": "online"}

@server.resource_template("users://{id}/profile", "User Profile", "Profile by ID", "application/json")
def profile(id: int) -> dict:
    return {"id": id, "role": "developer"}

@server.prompt("greet_user", "Generates a greeting prompt.")
def greet(username: str) -> str:
    return f"Hello {username}, how can I help you today?"

server.run(host="127.0.0.1", port=5000)   # → http://127.0.0.1:5000/mcp

from py_http_mcp_server import mcp_handler, mcp_tool, mcp_icon, mcp_resource, mcp_prompt

@mcp_handler
class MyAiTools:
    def __init__(self):            # register(MyAiTools) instantiates with no args;
        self.db = ...              # use register(MyAiTools(db)) to inject dependencies

    @mcp_tool("lookup", "Looks up a record.", title="Lookup", read_only_hint=True, icon_src="/icons/lookup.svg")
    @mcp_icon("/icons/lookup-dark.png", "image/png", "dark", "48x48")
    def lookup(self, key: str) -> dict: ...

server.register(MyAiTools)        # also accepts instances, modules, or decorated functions

from py_http_mcp_server import McpContent, McpToolResult

@server.tool("generate_chart", "Returns a chart image and resource link.")
def generate_chart(metric: str) -> list:
    return [
        McpContent.create_text(f"Generated chart for {metric}"),
        McpContent.create_image("iVBORw0KGgo...", "image/png"),
        McpContent.create_resource_link("system://status", "System Status", "Live status JSON", "application/json"),
    ]

from py_http_mcp_server import McpServer, OAuthProtectedResourceOptions, bearer_token_authenticator

server = McpServer(
    require_authorization=True,
    authenticator=bearer_token_authenticator({"secret-token": {"name": "McpUser"}}),  # or any callable(request) -> principal
    allowed_origins={"https://app.devin.ai"},
    oauth_protected_resource=OAuthProtectedResourceOptions(
        authorization_servers=["https://login.microsoftonline.com/<tenant-id>/v2.0"],
        scopes_supported=["api://my-mcp-server/mcp.tools"],
    ),
)

server.run(port=5000)                                  # blocking, Ctrl+C to stop
srv = server.start(port=0); print(srv.url); server.stop()  # background thread (tests)
server.run(port=443, ssl_context=my_ssl_context)       # HTTPS
application = server.wsgi_app                          # e.g. `waitress-serve module:application`
application = server.asgi_app                          # e.g. `uvicorn module:application`

server.notify_tools_list_changed()
server.broadcast("notifications/message", {"level": "info", "data": "hello"})

from py_http_mcp_server import HttpError, HttpRequest, SessionManager, sse_response

ui_events = SessionManager()

@server.route("/api/notes/{id}", methods=["GET", "DELETE"])
def note(request: HttpRequest):
    note_id = int(request.path_params["id"])        # also: request.query, request.json(), request.headers
    if note_id not in notes:
        raise HttpError(404, "Note not found")       # → 404 {"error": "Note not found"}
    return notes[note_id]                            # dict/list → JSON, str → HTML, None → 204, (body, 201)

@server.route("/api/events")                         # browser EventSource stream
def events(request: HttpRequest):
    return sse_response(ui_events)                   # push with ui_events.broadcast({...})

server.static("/", "web")                            # index.html, JS, CSS (path traversal is blocked)

@server.method("completion/complete")       # override a built-in or add your own
def complete(params: dict) -> dict:
    return {"completion": {"values": ["Seattle"], "total": 1, "hasMore": False}}

python -m unittest discover -s tests -t .