FastMCP Python 教程：只读工具、参数校验与访问控制测试

python3.11 -m venv .venv
.venv/bin/python -m pip install --only-binary=cryptography \
  'fastmcp==4.1.1' 'mcp==2.3.0' 'pydantic==2.14.1' \
  'PyJWT==2.15.1' 'cryptography==48.0.1'

from dataclasses import dataclass
from typing import Annotated

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.server.auth import AuthProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.server.dependencies import get_access_token
from mcp.types import ToolAnnotations
from pydantic import Field


@dataclass(frozen=True)
class Document:
    tenant: str
    doc_id: str
    title: str


# Synthetic data only. In production, apply tenant + ID in the database query.
DOCUMENTS = (
    Document("tenant-a", "a-001", "A team onboarding"),
    Document("tenant-b", "b-001", "B private handbook"),
)
# Trusted server-side mapping; never accept the tenant from tool arguments.
MEMBERSHIPS = {"alice": "tenant-a", "bob": "tenant-b"}


def current_tenant() -> str:
    token = get_access_token()
    if token is None or "docs:read" not in token.scopes:
        raise ToolError("Access denied")
    subject = token.claims.get("sub")
    if not isinstance(subject, str) or subject not in MEMBERSHIPS:
        raise ToolError("Access denied")
    return MEMBERSHIPS[subject]


def build_server(verifier: AuthProvider) -> FastMCP:
    server = FastMCP(
        "Read-only document demo",
        auth=verifier,
        strict_input_validation=True,
        mask_error_details=True,
    )

    @server.tool(
        annotations=ToolAnnotations(
            readOnlyHint=True,
            destructiveHint=False,
            idempotentHint=True,
            openWorldHint=False,
        )
    )
    def get_document(
        doc_id: Annotated[str, Field(
            min_length=5, max_length=5, pattern=r"^[ab]-[0-9]{3}$"
        )],
    ) -> dict[str, str]:
        """Read one document that belongs to the authenticated member."""
        tenant = current_tenant()
        doc = next((d for d in DOCUMENTS
                    if d.tenant == tenant and d.doc_id == doc_id), None)
        if doc is None:
            # Do not distinguish a foreign ID from a nonexistent ID.
            raise ToolError("Document unavailable")
        return {"doc_id": doc.doc_id, "title": doc.title}

    return server


if __name__ == "__main__":
    import os

    # Required configuration: a trusted IdP, not a URL supplied by the model.
    verifier = JWTVerifier(
        jwks_uri=os.environ["MCP_JWKS_URI"],
        issuer=os.environ["MCP_ISSUER"],
        audience=os.environ["MCP_AUDIENCE"],
        algorithm="RS256",
    )
    build_server(verifier).run(transport="http", host="127.0.0.1", port=8000)

import secrets
import time
import unittest
from unittest.mock import patch

import jwt
from fastmcp import Client
from fastmcp.exceptions import ToolError
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth.providers.jwt import JWTVerifier

import readonly_server as app


class ReadOnlyTests(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        # Ephemeral local test secret. Never reuse as a deployment credential.
        self.secret = secrets.token_urlsafe(48)
        self.verifier = JWTVerifier(
            public_key=self.secret, algorithm="HS256",
            issuer="test-issuer", audience="test-mcp",
        )
        self.server = app.build_server(self.verifier)

    def identity(self, subject="alice", scopes=None):
        return AccessToken(
            token="synthetic-test-only", client_id="test-client",
            scopes=["docs:read"] if scopes is None else scopes,
            claims={"sub": subject},
        )

    async def call(self, arguments, identity):
        # This substitutes verified identity ONLY. It does not test HTTP auth.
        with patch.object(app, "get_access_token", return_value=identity):
            async with Client(self.server) as client:
                return await client.call_tool("get_document", arguments)

    async def denied(self, arguments, identity):
        with self.assertRaises(ToolError) as caught:
            await self.call(arguments, identity)
        return str(caught.exception)

    async def test_schema_and_readonly_hint(self):
        async with Client(self.server) as client:
            tools = await client.list_tools()
        self.assertEqual([t.name for t in tools], ["get_document"])
        self.assertEqual(set(tools[0].input_schema["properties"]), {"doc_id"})
        self.assertTrue(tools[0].annotations.read_only_hint)
        self.assertFalse(tools[0].annotations.destructive_hint)

    async def test_own_document(self):
        result = await self.call({"doc_id": "a-001"}, self.identity())
        self.assertEqual(result.data,
                         {"doc_id": "a-001", "title": "A team onboarding"})

    async def test_cross_tenant_and_missing_share_error(self):
        foreign = await self.denied({"doc_id": "b-001"}, self.identity())
        missing = await self.denied({"doc_id": "a-999"}, self.identity())
        self.assertEqual(foreign, missing)
        self.assertNotIn("B private handbook", foreign)

    async def test_missing_identity(self):
        await self.denied({"doc_id": "a-001"}, None)

    async def test_missing_scope(self):
        await self.denied({"doc_id": "a-001"}, self.identity(scopes=[]))

    async def test_unknown_member(self):
        await self.denied({"doc_id": "a-001"}, self.identity(subject="eve"))

    async def test_invalid_inputs_and_tenant_injection(self):
        for args in [{}, {"doc_id": 1}, {"doc_id": "a-001-extra"},
                     {"doc_id": "../../x"}, {"doc_id": "c-001"},
                     {"doc_id": "a-001", "tenant_id": "tenant-b"}]:
            with self.subTest(args=args):
                await self.denied(args, self.identity())

    async def test_repeat_calls_leave_fixture_unchanged(self):
        before = tuple(app.DOCUMENTS)
        first = await self.call({"doc_id": "a-001"}, self.identity())
        second = await self.call({"doc_id": "a-001"}, self.identity())
        self.assertEqual(first.data, second.data)
        self.assertEqual(tuple(app.DOCUMENTS), before)

    async def test_native_jwt_verifier(self):
        now = int(time.time())
        claims = {"sub": "alice", "iss": "test-issuer", "aud": "test-mcp",
                  "exp": now + 300, "scope": "docs:read"}
        signed = jwt.encode(claims, self.secret, algorithm="HS256")
        verified = await self.verifier.verify_token(signed)
        self.assertIsNotNone(verified)
        self.assertEqual(verified.claims["sub"], "alice")
        self.assertIn("docs:read", verified.scopes)
        invalid = [jwt.encode(claims, secrets.token_urlsafe(48), algorithm="HS256")]
        for change in [{"iss": "other"}, {"aud": "other"}, {"exp": now - 300}]:
            invalid.append(jwt.encode(claims | change, self.secret, algorithm="HS256"))
        for value in invalid:
            with self.subTest(kind="invalid token"):
                self.assertIsNone(await self.verifier.verify_token(value))


if __name__ == "__main__":
    unittest.main(verbosity=2)

.venv/bin/python -m unittest -v test_readonly

Ran 9 tests
OK