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:
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.
This standalone example deliberately uses deterministic checks so it runs without a model. The main product policy enables Laya by default. To enable the same semantic input and output checks in this isolated example, follow Enable Laya below.
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:4b
threshold: 0.7
timeout_ms: 10000
scan_output: true
tools:
text.uppercase:
roles:
- analyst
timeout_ms: 5000
models:
qwen3:4b:
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
Enable Laya in this example¶
Stop the example server. From your main installation directory, with Ollama running, prepare the runtime:
uv tool run --python 3.12 [email protected] init --anonymization
export FASTFENCE_AUTHORING_ROOT="$PWD"
The environment variable lets the isolated example use the main installation's Laya engine. If you changed the Ollama endpoint, also export the same FASTFENCE_OLLAMA_URL in this shell; the example reads environment variables, not the main installation's .env file.
Edit state/examples/fastmcp-integration/config/policy.yaml, which was created on the example's first start. Preserve its tools, budgets and other controls, increment its current top-level version, and replace its semantic section with:
Use the assessment model prepared by your main installation if you changed it from Qwen3:4b. Installing Laya alone does not enable assessment: provider: laya in this example's own policy is required.
Restart from the same shell:
Send hello again. For an allowed response, both semantic_input_status and semantic_output_status should be passed. Missing or failed assessment blocks the request. Input denied by an earlier local rule never reaches the assessor or tool.
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.