starter-mcp-server/
├── .github/                         # CI/CD workflows
│   └── workflows/
│       ├── pull-request-check.yaml
│       ├── build-publish.yaml
│       ├── release-draft.yaml
│       ├── release.yaml
│       └── repository-provisioning.yaml
├── src/
│   ├── main.py                      # Entry point
│   └── starter_mcp_server.py        # MCP server (config, OpenAPI, tools, structured output, elicitation)
├── tests/
│   ├── test_config.py
│   ├── test_header_forwarding.py
│   ├── test_main.py
│   ├── test_server.py
│   └── test_tools.py
├── spec/
│   └── platzi-fake-store-api.yaml   # Sample OpenAPI spec
├── .env.example                     # Environment template
├── Dockerfile                       # Container definition
└── pyproject.toml                   # Dependencies

from fastmcp import FastMCP

mcp = FastMCP(name="MyServer")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

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

from typing import Annotated
from pydantic import Field
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP(name="CalculatorServer")

@mcp.tool(
    name="find_products",
    description="Search the product catalog with optional category filtering.",
    tags={"catalog", "search"},
    timeout=30.0,
    annotations={"readOnlyHint": True},
)
def search_products(query: str, category: str | None = None) -> list[dict]:
    return [{"id": 2, "name": "Widget"}]

@mcp.tool
def process_image(
    image_url: Annotated[str, "URL of the image to process"],
    width: Annotated[int, Field(description="Target width", ge=1, le=2000)] = 800,
) -> dict:
    """Process an image with optional resizing."""
    ...

@mcp.tool
def divide(a: float, b: float) -> float:
    """Divide a by b."""
    if b == 0:
        raise ToolError("Division by zero is not allowed.")
    return a / b

from fastmcp import FastMCP, Context
from fastmcp.server.elicitation import (
    AcceptedElicitation,
    DeclinedElicitation,
    CancelledElicitation,
)
from pydantic import BaseModel, Field

mcp = FastMCP(name="OrdersServer")

class OrderDetails(BaseModel):
    confirm: bool = Field(description="Whether to proceed with the order")
    quantity: int = Field(default=1, ge=1, le=100)
    shipping_address: str = Field(default="")

@mcp.tool
async def order_product(ctx: Context, product_id: int) -> dict:
    """Place an order after asking the user to confirm."""
    result = await ctx.elicit(
        message="Please confirm your order.",
        response_type=OrderDetails,
    )

    # Handle every response branch
    if isinstance(result, CancelledElicitation):
        return {"status": "cancelled"}
    if isinstance(result, DeclinedElicitation):
        return {"status": "declined"}

    order = result.data  # OrderDetails, available on AcceptedElicitation
    if not order.confirm:
        return {"status": "not_confirmed"}
    return {"status": "success", "quantity": order.quantity}

import httpx
import yaml
from fastmcp import FastMCP

with open("my-api-spec.yaml") as f:
    openapi_spec = yaml.safe_load(f)

client = httpx.AsyncClient(base_url="https://api.example.com")

mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    client=client,
    name="My API Server",
)

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

import httpx
from fastmcp import FastMCP

mcp = FastMCP("API Tools Server")

@mcp.tool
async def get_payment_orders(status: str | None = None) -> dict:
    """Retrieve payment orders, optionally filtered by status."""
    async with httpx.AsyncClient() as client:
        params = {}
        if status:
            params["status"] = status
        response = await client.get(
            "https://api.example.com/client-api/v3/payment-orders",
            params=params,
            headers={"Authorization": "Bearer TOKEN"},
        )
        response.raise_for_status()
        return response.json()

from typing import Any
import httpx
from fastmcp.server.dependencies import get_http_request

class HeaderForwardingClient(httpx.AsyncClient):
    """Forwards auth headers from the MCP client request to the upstream API."""

    async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response:
        try:
            incoming = get_http_request()
            auth = incoming.headers.get("Authorization", "")
            if auth:
                request.headers["Authorization"] = auth
        except Exception:
            pass
        request.headers["Content-Type"] = "application/json"
        return await super().send(request, **kwargs)

api_client = HeaderForwardingClient(base_url="https://api.example.com")

mcp = FastMCP.from_openapi(
    openapi_spec=spec,
    client=api_client,
    name="My Authenticated API Server",
)

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",  # "stdio" (default) | "streamable-http"
        host="0.0.0.0",
        port=8557,
        path="/mcp",
        stateless_http=False,  # keep stateful for elicitation; set True for stateless deployments
    )

from starlette.requests import Request
from starlette.responses import JSONResponse

@mcp.custom_route("/health", methods=["GET"])
async def health(request: Request) -> JSONResponse:
    return JSONResponse({"status": "ok"})

uv sync --extra dev
uv run pytest

docker build -t starter-mcp-server:local .