Skip to content

Protect a FastMCP server inside a FastAPI application

This complete application connects a real FastMCP uppercase tool to FastFence through ToolsPort. FastFence's FastAPI application exposes authenticated REST and MCP entry points. Its input controls run before the tool, and output controls run before delivery.

The private FastMCP backend is in-process and has no unprotected listening port. This avoids publishing a second route that bypasses the gateway. For a remote MCP backend, replace Client(backend) with a client for a fixed trusted URL, supply its separate server-side credential, and restrict direct access to that backend.

Run

After installing the package, extract the examples archive into examples/ in your installation directory. Keep policy.yaml and signatures.json next to fastmcp_server.py. Then run:

python examples/fastmcp_server.py

The application listens on http://127.0.0.1:8010. It initializes a separate policy and credentials in state/examples/fastmcp-integration/; it does not change the main installation. Open this console and connect the local-agent and local-admin credentials from that directory's state/credentials.json.

The example uses deterministic checks so it runs without a model. This is an explicit example profile; the product's default policy enables Laya. To add semantic checking here, set up Laya for this installation and configure its trusted authoring_root as described in installation.

Complete server and FastAPI integration

"""Expose a private FastMCP tool through FastFence's FastAPI and MCP interfaces."""

import json
import shutil
from pathlib import Path
from typing import Any

import uvicorn
from fastmcp import Client, FastMCP
from pydantic import BaseModel, ConfigDict, Field

from fastfence.app.factory import create_app
from fastfence.app.interfaces.cli.initialize import initialize
from fastfence.modules.control.contracts.dto import Identity
from fastfence.shared.settings.app_settings import AppSettings

backend = FastMCP("Private text tools")


@backend.tool()
def uppercase(text: str) -> dict[str, str]:
    """An actual, deterministic operation; replace with your business logic."""
    return {"text": text.upper()}


class UppercaseInput(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    text: str = Field(min_length=1, max_length=1024)


class ProtectedMCPTools:
    def supports(self, tool: str) -> bool:
        return tool == "text.uppercase"

    def validate(
        self, tool: str, arguments: dict[str, Any], identity: Identity
    ) -> dict[str, Any]:
        if not self.supports(tool):
            raise ValueError("Unknown tool")
        return UppercaseInput.model_validate(arguments).model_dump()

    async def call(
        self, tool: str, arguments: dict[str, Any], identity: Identity
    ) -> dict[str, Any]:
        # FastFence calls this only after authentication and input controls.
        if not self.supports(tool):
            raise ValueError("Unknown tool")
        async with Client(backend) as client:
            result = await client.call_tool("uppercase", arguments)
        if not isinstance(result.data, dict):
            raise ValueError("Unexpected MCP output")
        return result.data


def build_example(root: Path):
    """Use a separate config/state directory; never edit the operator's policy."""
    config = root / "config"
    config.mkdir(parents=True, exist_ok=True)
    for name in ("policy.yaml", "signatures.json"):
        target = config / name
        if not target.exists():
            shutil.copyfile(Path(__file__).with_name(name), target)
    initialize(root / "state")
    app = create_app(
        AppSettings(root=root, state=root / "state"), tools=ProtectedMCPTools()
    )

    @app.get("/integration-info")
    def integration_info():
        return {"tool": "text.uppercase", "mcp": "/mcp/", "rest": "/api/invoke"}

    return app


def main() -> None:
    root = Path("state/examples/fastmcp-integration").resolve()
    app = build_example(root)
    # Print paths only; keep provisioned bearer credentials private.
    print(json.dumps({"url": "http://127.0.0.1:8010", "state": str(root)}))
    uvicorn.run(app, host="127.0.0.1", port=8010)


if __name__ == "__main__":
    main()

Download fastmcp_server.py · View source

create_app(..., tools=ProtectedMCPTools()) is the integration point. Register business operations through this port; an ordinary FastAPI route is not automatically protected by FastFence. The public /integration-info route returns static metadata only. The verified identity passed to the adapter can also enforce application-specific tenant ownership before execution.

Policy

version: 1
description: Strict input protection with useful, redacted output
privacy:
  enabled: true
  input: block
  output: redact
signatures_enabled: true
max_input_bytes: 16384
max_output_bytes: 16384
semantic:
  provider: disabled
  model: qwen3:0.6b
  threshold: 0.7
  timeout_ms: 10000
  scan_output: true
tools:
  text.uppercase:
    roles:
    - analyst
    timeout_ms: 5000
models:
  qwen3:0.6b:
    roles:
    - analyst
    - operator
    max_output_tokens: 256
    timeout_ms: 30000
    cost_microusd: 0
budgets:
  analyst:
    calls: 20
    tokens: 1000000
    cost_microusd: 10000
    compute_ms: 180000
    concurrent: 4
  operator:
    calls: 30
    tokens: 2000000
    cost_microusd: 20000
    compute_ms: 300000
    concurrent: 4
text_rules:
- id: forbidden-word
  operator: contains
  value: forbidden
  direction: input
  target: tool
  case_sensitive: false

Download policy.yaml · View source

Invoke through REST

Set FASTFENCE_AGENT_TOKEN to the example's provisioned agent token in your shell, then:

curl http://127.0.0.1:8010/api/invoke \
  -H "Authorization: Bearer $FASTFENCE_AGENT_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"tool":"text.uppercase","arguments":{"text":"hello"}}'

Expected: decision: allowed, upstream_executed: true, and output.text: HELLO.

Repeat with {"text":"forbidden"}. Expected: decision: blocked, upstream_executed: false. FastFence blocks the exact sample word before calling FastMCP. The same controls apply through the FastMCP client, using this server's port and tool identifier.

For an output-only test, add a rule matching HELLO, direction output, target tool, case sensitive. The tool executes, but its response is withheld. Inspect Activity to distinguish input and output blocks. HTTP 200 by itself never means the operation was allowed.