Chroń serwer FastMCP wewnątrz aplikacji FastAPI¶
Ta kompletna aplikacja łączy rzeczywiste narzędzie FastMCP uppercase z FastFence przez ToolsPort. Aplikacja FastAPI w FastFence udostępnia uwierzytelnione wejścia REST i MCP. Kontrole wejścia wykonują się przed narzędziem, a kontrole wyjścia przed dostarczeniem wyniku.
Prywatny backend FastMCP działa w tym samym procesie i nie nasłuchuje na niezabezpieczonym porcie. Nie powstaje dzięki temu druga trasa omijająca bramkę. Dla zdalnego backendu MCP zastąp Client(backend) klientem o stałym zaufanym adresie, przekaż osobny token po stronie serwera i ogranicz bezpośredni dostęp do backendu.
Uruchom¶
Po zainstalowaniu pakietu rozpakuj archiwum przykładów do examples/ w katalogu instalacji. Zachowaj policy.yaml i signatures.json obok fastmcp_server.py. Następnie uruchom:
Aplikacja nasłuchuje pod http://127.0.0.1:8010. Inicjalizuje osobną politykę i tokeny w state/examples/fastmcp-integration/; nie zmienia głównej instalacji. Otwórz tę konsolę i połącz tokeny local-agent oraz local-admin z pliku state/credentials.json w tym katalogu.
Ten samodzielny przykład celowo korzysta z kontroli deterministycznych, aby działał bez modelu. Główna polityka produktu domyślnie włącza Laya. Aby włączyć te same kontrole semantyczne wejścia i wyjścia w odizolowanym przykładzie, wykonaj poniższe kroki Włącz Laya.
Kompletny serwer i integracja z FastAPI¶
"""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()
Pobierz fastmcp_server.py · Zobacz źródło
Punktem integracji jest create_app(..., tools=ProtectedMCPTools()). Rejestruj operacje biznesowe przez ten port; zwykła trasa FastAPI nie jest automatycznie chroniona przez FastFence. Publiczna trasa /integration-info zwraca wyłącznie statyczne metadane. Zweryfikowana tożsamość przekazana adapterowi może też służyć do sprawdzania własności zasobów tenanta przed wykonaniem operacji.
Polityka¶
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
Pobierz policy.yaml · Zobacz źródło
Włącz Laya w tym przykładzie¶
Zatrzymaj serwer przykładu. W głównym katalogu instalacji, przy działającym Ollama przygotuj środowisko wykonawcze:
uv tool run --python 3.12 [email protected] init --anonymization
export FASTFENCE_AUTHORING_ROOT="$PWD"
Zmienna środowiskowa pozwala odizolowanemu przykładowi korzystać z silnika Laya głównej instalacji. Jeśli zmieniłeś endpoint Ollama, wyeksportuj w tej powłoce również takie samo FASTFENCE_OLLAMA_URL; przykład czyta zmienne środowiskowe, a nie plik .env głównej instalacji.
Edytuj state/examples/fastmcp-integration/config/policy.yaml, utworzony przy pierwszym starcie przykładu. Zachowaj narzędzia, budżety i pozostałe kontrole, zwiększ bieżącą wersję version na najwyższym poziomie i zastąp sekcję semantic następującą:
Jeśli zmieniłeś model oceniający z Qwen3:4b, użyj modelu przygotowanego w głównej instalacji. Sama instalacja Laya nie włącza oceniania: polityka tego przykładu musi zawierać provider: laya.
Uruchom ponownie z tej samej powłoki:
Wyślij ponownie hello. Przy dozwolonej odpowiedzi zarówno semantic_input_status, jak i semantic_output_status powinny mieć wartość passed. Brak oceny lub błąd jej wykonania blokuje żądanie. Wejście odrzucone przez wcześniejszą regułę lokalną nigdy nie dociera do modelu oceniającego ani do narzędzia.
Wywołaj przez REST¶
Ustaw w powłoce FASTFENCE_AGENT_TOKEN na wygenerowany token agenta tego przykładu, a następnie:
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"}}'
Oczekiwany wynik: decision: allowed, upstream_executed: true i output.text: HELLO.
Powtórz z {"text":"forbidden"}. Oczekuj decision: blocked i upstream_executed: false. FastFence blokuje dokładne przykładowe słowo przed wywołaniem FastMCP. Te same kontrole obowiązują przez klienta FastMCP, z portem tego serwera i identyfikatorem narzędzia.
Aby sprawdzić samo wyjście, dodaj regułę dopasowującą HELLO, kierunek output, cel tool, z rozróżnianiem wielkości liter. Narzędzie wykona się, ale odpowiedź zostanie zatrzymana. W Activity rozróżnisz blokadę wejścia od blokady wyjścia. Sam HTTP 200 nigdy nie oznacza zezwolenia na operację.