lane-mcp-auth

Lane sign-in for an MCP server, in Python. The TypeScript distribution is @getonlane/mcp-auth.

What it does for your MCP server

• Lane sign-in for agents. Lane is your OAuth authorization server. The library verifies each bearer, serves the protected-resource document, and refuses every tool until the agent registers a session with your server.
• Scopes, guest tools and step-up. A tool names the scopes it needs. A guest calls your public read tools with no Lane account. A missing scope starts a new Lane sign-in.
• Purchases that a human approves. A priced tool answers with a proposal. The user approves the plan in their Lane wallet. Lane then calls the tool with an execution grant, and the library consumes each grant once.
• Plan step events. Lane tells your server what happened to each step it sold, so your server fulfils, frees stock or cancels.
• Telemetry. Lane records sign-ins and tool calls for your server, and on_gate_event gives you each gate decision in your own process.

Install

pip install "lane-mcp-auth[fastmcp]"
python -m lane_mcp_auth setup https://acme.com/mcp --name="Acme shop"

setup pairs your terminal with the Lane console and writes LANE_CLIENT_ID, LANE_CLIENT_SECRET, LANE_VERIFICATION, LANE_RESOURCE and LANE_ISSUER to .env. When Lane mints a sandbox org key for the pairing, it also writes LANE_ORG_KEY. The library requires an org key, and its prefix sets the environment: lane_org_sk_test_ is sandbox. With an org key, setup registers the server with no browser. It takes the key from --org-key, then from LANE_ORG_KEY in .env, then from the process environment. It writes the key it used to .env as LANE_ORG_KEY. The [fastmcp] extra adds only the FastMCP adapter.

Example

One gated tool on FastMCP 4, served over HTTP. The store keeps connections in memory, so it is correct for one process only.

import os

from fastmcp import Context, FastMCP
from fastmcp.server.dependencies import get_http_request
from fastmcp.server.middleware import Middleware
from lane_mcp_auth import ConnectionKey, ConnectionRecord, HttpTokenExchanger, LaneMcpAuth
from lane_mcp_auth.fastmcp import CLAIMS_ATTR, enable_lane_auth, requires
from starlette.middleware import Middleware as AsgiMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class MemoryConnections:
    def __init__(self) -> None:
        self.rows: dict[tuple[str, str], ConnectionRecord] = {}

    async def get(self, key: ConnectionKey) -> ConnectionRecord | None:
        return self.rows.get((key.sub, key.client_id))

    async def put(self, key: ConnectionKey, record: ConnectionRecord) -> ConnectionRecord:
        self.rows[(key.sub, key.client_id)] = record
        return record


auth = LaneMcpAuth(
    resource=os.environ["LANE_RESOURCE"],
    client_id=os.environ["LANE_CLIENT_ID"],
    announce_secret=os.environ["LANE_CLIENT_SECRET"],
    verification=os.environ.get("LANE_VERIFICATION"),
    org_key=os.environ["LANE_ORG_KEY"],
    connections=MemoryConnections(),
    exchanger=HttpTokenExchanger(
        client_id=os.environ["LANE_CLIENT_ID"],
        client_secret=os.environ["LANE_CLIENT_SECRET"],
    ),
)

mcp = FastMCP("acme")


class AttachLaneClaims(Middleware):
    async def on_call_tool(self, context, call_next):
        claims = get_http_request().state.lane_claims
        setattr(context.fastmcp_context.request_context, CLAIMS_ATTR, claims)
        return await call_next(context)


class VerifyLaneBearer(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path.startswith("/.well-known/"):
            return await call_next(request)
        token = request.headers.get("authorization", "").removeprefix("Bearer ").strip() or None
        claims = auth.authenticate(token)
        if claims is None:
            return Response(status_code=401, headers={"WWW-Authenticate": auth.challenge(token)})
        request.state.lane_claims = claims
        return await call_next(request)


mcp.add_middleware(AttachLaneClaims())
enable_lane_auth(mcp, auth)


@mcp.tool()
@requires("read")
async def order_history(ctx: Context) -> str:
    return "two orders"


for path in filter(None, auth.metadata_paths()):
    @mcp.custom_route(path, methods=["GET"])
    async def lane_document(request) -> Response:
        return Response(auth.protected_resource_document(), media_type="application/json")


app = mcp.http_app(middleware=[AsgiMiddleware(VerifyLaneBearer)])

Serve app with any ASGI server, for example uvicorn server:app. An agent that calls order_history is told to call lane_register_session first. After that call, the tool runs when the connection holds <your host>/read.

Documentation

Start with the quickstart.

• Overview: how the parts fit
• Connections: the store you supply, and what to run in production
• Scopes: Lane scopes, your scopes, guests and step-up
• Customize: staged rollout, guests, revocation, identity
• Charging for a purchase and plan step events
• Telemetry
• Auth library reference: every export, and where Python and TypeScript differ

Licence

MIT