diff --git a/packages/fastmcp/README.md b/packages/fastmcp/README.md index f67e51b4..0309a902 100644 --- a/packages/fastmcp/README.md +++ b/packages/fastmcp/README.md @@ -16,8 +16,8 @@ pip install keycardai-fastmcp ## Quick Start ```python -from fastmcp import FastMCP, Context -from keycardai.fastmcp import AuthProvider +from fastmcp import FastMCP +from keycardai.fastmcp import AccessContext, AuthProvider auth_provider = AuthProvider( zone_id="abc1234", @@ -28,13 +28,65 @@ auth_provider = AuthProvider( mcp = FastMCP("My Server", auth=auth_provider.get_remote_auth_provider()) @mcp.tool() -@auth_provider.grant("https://api.example.com") -async def call_external_api(ctx: Context, query: str): - keycardai = await ctx.get_state("keycardai") - token = keycardai.access("https://api.example.com").access_token +async def call_external_api( + query: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), +): + token = access.access("https://api.example.com").access_token return f"Results for {query} (token starts with {token[:8]})" ``` +Declaring the grant as a typed parameter default injects the populated +`AccessContext` per request; the parameter never appears in the tool's input +schema. Exchange failures are recorded on the `AccessContext` (check +`access.has_errors()` / `access.get_errors()`), never raised. Granting +multiple resources is all-or-nothing: if any exchange fails, the context +carries that resource's error and no tokens. + +If you lint with flake8-bugbear or Ruff's `B008` rule (function call in +argument default), exempt your tool modules: the call-in-default is the +intended spelling here, the same pattern as FastAPI's `Depends`. In +`pyproject.toml`: + +```toml +[tool.ruff.lint.per-file-ignores] +"src/my_server/tools/*.py" = ["B008"] +``` + +### Migrating from the decorator form + +The decorator form (`@auth_provider.grant(...)` above the tool) still works +from the same object. Reading the result via `ctx.get_state("keycardai")` is +deprecated and emits a `DeprecationWarning`; helpers that only hold the +FastMCP `Context` can use `await AccessContext.from_context(ctx)` instead. + +The warning fires once per tool, at decoration time (module import). If your +test or CI setup escalates warnings to errors (`-W error`, +`filterwarnings = ["error"]` in pytest config), importing a server module +that still uses the old form will raise instead of warn. Either migrate the +tools to the injected-parameter form, or allow this warning explicitly: + +```toml +filterwarnings = ["error", 'default:Tool .* uses the grant decorator:DeprecationWarning'] +``` + +## Testing + +Fake delegated access without patching internals: + +```python +from keycardai.fastmcp.testing import mock_access_context + +with mock_access_context(access_token="fake_token"): + ... # grants resolve to an AccessContext serving fake_token +``` + +The bare `access_token` form serves the token for any resource, so it will +not catch a mistyped resource URL in an `access(...)` call. Pass +`resource_tokens={...}` when the test should enforce which resources the +tool reads; resources outside the dict raise `ResourceAccessError`, matching +production. + ## Migration from `keycardai-mcp-fastmcp` The old package keeps working: `from keycardai.mcp.integrations.fastmcp import AuthProvider` diff --git a/packages/fastmcp/examples/delegated_access/README.md b/packages/fastmcp/examples/delegated_access/README.md index 3291bc3c..7cc6436e 100644 --- a/packages/fastmcp/examples/delegated_access/README.md +++ b/packages/fastmcp/examples/delegated_access/README.md @@ -1,6 +1,6 @@ # GitHub API Integration with Keycard Delegated Access -A complete example demonstrating how to use the `@grant` decorator for token exchange, enabling your MCP server to access external APIs (GitHub) on behalf of authenticated users. +A complete example demonstrating how to declare a grant as a typed tool parameter for token exchange, enabling your MCP server to access external APIs (GitHub) on behalf of authenticated users. ## Why Keycard? @@ -150,7 +150,7 @@ User MCP Server Keycard GitHub ``` 1. User authenticates to your MCP server via Keycard -2. When a tool with `@grant` is called, Keycard exchanges the user's token +2. When a tool declaring a grant parameter is called, Keycard exchanges the user's token 3. The exchanged token has the scopes configured for the external resource 4. Your server uses this token to call GitHub API on behalf of the user diff --git a/packages/fastmcp/examples/delegated_access/main.py b/packages/fastmcp/examples/delegated_access/main.py index 4f6b3cda..e4f2fd62 100644 --- a/packages/fastmcp/examples/delegated_access/main.py +++ b/packages/fastmcp/examples/delegated_access/main.py @@ -1,20 +1,20 @@ """GitHub API Integration with Keycard Delegated Access. -This example demonstrates how to use the @grant decorator to request -token exchange for accessing external APIs (GitHub) on behalf of +This example demonstrates how to declare a grant as a typed tool parameter +to request token exchange for accessing external APIs (GitHub) on behalf of authenticated users. Key concepts demonstrated: - AuthProvider setup with ClientSecret credentials -- @grant decorator for requesting token exchange -- AccessContext for accessing exchanged tokens +- auth_provider.grant(...) as a typed parameter default for token exchange +- AccessContext injected into the tool, hidden from the tool's input schema - Comprehensive error handling patterns """ import os import httpx -from fastmcp import Context, FastMCP +from fastmcp import FastMCP from keycardai.fastmcp import AccessContext, AuthProvider, ClientSecret @@ -41,31 +41,29 @@ @mcp.tool() -@auth_provider.grant("https://api.github.com") -async def get_github_user(ctx: Context) -> dict: +async def get_github_user( + access: AccessContext = auth_provider.grant("https://api.github.com"), +) -> dict: """Get the authenticated GitHub user's profile. Demonstrates: - - Basic @grant decorator usage + - Basic grant-as-parameter usage - Error checking with has_errors() - - Token access via AccessContext + - Token access via the injected AccessContext Args: - ctx: FastMCP context with Keycard authentication state + access: Injected access context holding the exchanged GitHub token Returns: User profile data or error details """ - # Get access context from FastMCP context namespace - access_context: AccessContext = await ctx.get_state("keycardai") - # Check for any errors (global or resource-specific) - if access_context.has_errors(): - errors = access_context.get_errors() + if access.has_errors(): + errors = access.get_errors() return {"error": "Token exchange failed", "details": errors} # Get the exchanged token for GitHub API - token = access_context.access("https://api.github.com").access_token + token = access.access("https://api.github.com").access_token # Call GitHub API with delegated token async with httpx.AsyncClient() as client: @@ -94,37 +92,37 @@ async def get_github_user(ctx: Context) -> dict: @mcp.tool() -@auth_provider.grant("https://api.github.com") -async def list_github_repos(ctx: Context, per_page: int = 5) -> dict: +async def list_github_repos( + per_page: int = 5, + access: AccessContext = auth_provider.grant("https://api.github.com"), +) -> dict: """List the authenticated user's GitHub repositories. Demonstrates: - Resource-specific error checking with has_resource_error() - Getting resource-specific errors with get_resource_error() - - Parameterized API calls + - Mixing regular tool parameters with an injected grant Args: - ctx: FastMCP context with Keycard authentication state per_page: Number of repositories to return (default: 5) + access: Injected access context holding the exchanged GitHub token Returns: List of repositories or error details """ - access_context: AccessContext = await ctx.get_state("keycardai") - # Check for resource-specific error (alternative to has_errors()) - if access_context.has_resource_error("https://api.github.com"): - resource_errors = access_context.get_resource_error("https://api.github.com") + if access.has_resource_error("https://api.github.com"): + resource_errors = access.get_resource_error("https://api.github.com") return { "message": "Token exchange failed for GitHub API", "details": resource_errors, } # Check for global errors (e.g., no auth token available) - if access_context.has_error(): - return {"error": "Global token error", "details": access_context.get_error()} + if access.has_error(): + return {"error": "Global token error", "details": access.get_error()} - token = access_context.access("https://api.github.com").access_token + token = access.access("https://api.github.com").access_token async with httpx.AsyncClient() as client: response = await client.get( diff --git a/packages/fastmcp/examples/delegated_access/pyproject.toml b/packages/fastmcp/examples/delegated_access/pyproject.toml index a3f12af8..eee70372 100644 --- a/packages/fastmcp/examples/delegated_access/pyproject.toml +++ b/packages/fastmcp/examples/delegated_access/pyproject.toml @@ -6,7 +6,7 @@ readme = "README.md" requires-python = ">=3.10" dependencies = [ "keycardai-fastmcp", - "fastmcp>=3.0.0", + "fastmcp>=3.1.0", "httpx>=0.27.0,<1.0.0", ] diff --git a/packages/fastmcp/examples/hello_world_server/README.md b/packages/fastmcp/examples/hello_world_server/README.md index 152148f3..82f94cf1 100644 --- a/packages/fastmcp/examples/hello_world_server/README.md +++ b/packages/fastmcp/examples/hello_world_server/README.md @@ -57,7 +57,7 @@ Connect to your server using any MCP-compatible client (e.g., Cursor, Claude Des ## Delegated Access -For accessing external APIs on behalf of users using the `@grant` decorator, see the [Delegated Access Example](../delegated_access/). +For accessing external APIs on behalf of users with an injected grant parameter, see the [Delegated Access Example](../delegated_access/). ## Environment Variables Reference diff --git a/packages/fastmcp/examples/hello_world_server/pyproject.toml b/packages/fastmcp/examples/hello_world_server/pyproject.toml index 8af4f11f..8a7ab0f7 100644 --- a/packages/fastmcp/examples/hello_world_server/pyproject.toml +++ b/packages/fastmcp/examples/hello_world_server/pyproject.toml @@ -6,7 +6,7 @@ readme = "README.md" requires-python = ">=3.10" dependencies = [ "keycardai-fastmcp", - "fastmcp>=3.0.0", + "fastmcp>=3.1.0", ] [tool.uv.sources] diff --git a/packages/fastmcp/pyproject.toml b/packages/fastmcp/pyproject.toml index 64b461d2..70032d6d 100644 --- a/packages/fastmcp/pyproject.toml +++ b/packages/fastmcp/pyproject.toml @@ -11,7 +11,7 @@ dependencies = [ "pydantic-settings>=2.7.1", "httpx>=0.27.2", "keycardai-oauth>=0.7.0", - "fastmcp>=3.0.0", + "fastmcp>=3.1.0", "keycardai-mcp>=0.15.0", ] keywords = ["fastmcp", "mcp", "model-context-protocol", "oauth", "token-exchange", "authentication", "keycard"] @@ -84,7 +84,11 @@ ignore = [ isort = { combine-as-imports = true, known-first-party = ["keycardai"] } [tool.ruff.lint.per-file-ignores] -"tests/**/*.py" = ["T20"] +# B008: auth_provider.grant(...) is a dependency declaration (like +# fastapi.Depends): calling it in a parameter default is the intended +# injection spelling for tools. +"tests/**/*.py" = ["T20", "B008"] +"examples/**/*.py" = ["B008"] [tool.mypy] strict = true diff --git a/packages/fastmcp/src/keycardai/fastmcp/__init__.py b/packages/fastmcp/src/keycardai/fastmcp/__init__.py index e1ee6451..f88113a4 100644 --- a/packages/fastmcp/src/keycardai/fastmcp/__init__.py +++ b/packages/fastmcp/src/keycardai/fastmcp/__init__.py @@ -4,11 +4,16 @@ and FastMCP servers, enabling secure authentication and authorization. Components: -- AuthProvider: Keycard authentication provider with RemoteAuthProvider creation and grant decorator -- AccessContext: Context object for accessing delegated tokens (used in FastMCP Context namespace) +- AuthProvider: Keycard authentication provider with RemoteAuthProvider creation and grant dependency +- AccessContext: Typed context object for accessing delegated tokens, injected into tool parameters - Application credentials: ClientSecret, WebIdentity, EKSWorkloadIdentity for different authentication scenarios - Auth strategies: BasicAuth, MultiZoneBasicAuth, NoneAuth for HTTP client authentication +GrantDependency is the return type of AuthProvider.grant(); it is exported for +type annotations only and is never constructed directly. Testing seams live in +keycardai.fastmcp.testing (mock_access_context is also re-exported here for +backward compatibility). + Re-export Guide: Local definitions (primary API): AuthProvider, AccessContext From keycardai.mcp.server.auth: ApplicationCredential, ClientSecret, EKSWorkloadIdentity, WebIdentity @@ -19,8 +24,8 @@ Basic Usage: - from fastmcp import FastMCP, Context - from keycardai.fastmcp import AuthProvider + from fastmcp import FastMCP + from keycardai.fastmcp import AccessContext, AuthProvider # Create authentication provider auth_provider = AuthProvider( @@ -33,11 +38,13 @@ auth = auth_provider.get_remote_auth_provider() mcp = FastMCP("My Server", auth=auth) - # Use grant decorator for token exchange + # Declare the grant as a typed tool parameter for token exchange @mcp.tool() - @auth_provider.grant("https://api.example.com") - async def call_external_api(ctx: Context, query: str): - token = (await ctx.get_state("keycardai")).access("https://api.example.com").access_token + async def call_external_api( + query: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ): + token = access.access("https://api.example.com").access_token # Use token to call external API return f"Results for {query}" @@ -55,11 +62,14 @@ async def call_external_api(ctx: Context, query: str): # Multiple resource access @mcp.tool() - @auth_provider.grant(["https://www.googleapis.com/calendar/v3", "https://www.googleapis.com/drive/v3"]) - async def sync_calendar_to_drive(ctx: Context): - access_context = await ctx.get_state("keycardai") - calendar_token = access_context.access("https://www.googleapis.com/calendar/v3").access_token - drive_token = access_context.access("https://www.googleapis.com/drive/v3").access_token + async def sync_calendar_to_drive( + access: AccessContext = auth_provider.grant([ + "https://www.googleapis.com/calendar/v3", + "https://www.googleapis.com/drive/v3", + ]), + ): + calendar_token = access.access("https://www.googleapis.com/calendar/v3").access_token + drive_token = access.access("https://www.googleapis.com/drive/v3").access_token # Use both tokens for cross-service operations return "Sync completed" @@ -107,13 +117,20 @@ async def sync_calendar_to_drive(ctx: Context): NoneAuth, ) -from .provider import AccessContext, AuthProvider +from .provider import ( + AccessContext, + AuthProvider, + GrantDependency, +) from .testing import mock_access_context __all__ = [ # === Primary API (Local Definitions) === "AuthProvider", "AccessContext", + # === Typing Support === + # Return type of AuthProvider.grant(); exported for annotations, never constructed directly + "GrantDependency", # === Application Credentials (re-exported from keycardai.mcp.server.auth) === "ApplicationCredential", "ClientSecret", diff --git a/packages/fastmcp/src/keycardai/fastmcp/provider.py b/packages/fastmcp/src/keycardai/fastmcp/provider.py index bb8e0770..851416ca 100644 --- a/packages/fastmcp/src/keycardai/fastmcp/provider.py +++ b/packages/fastmcp/src/keycardai/fastmcp/provider.py @@ -11,17 +11,22 @@ import inspect import logging import os +import warnings from collections.abc import Callable +from contextlib import contextmanager +from contextvars import ContextVar from functools import wraps -from typing import Any +from types import UnionType +from typing import Annotated, Any, Union, get_args, get_origin, get_type_hints from urllib.parse import urlparse from pydantic import AnyHttpUrl from fastmcp import Context +from fastmcp.dependencies import Dependency from fastmcp.server.auth import RemoteAuthProvider from fastmcp.server.auth.providers.jwt import JWTVerifier -from fastmcp.server.dependencies import get_access_token +from fastmcp.server.dependencies import get_access_token, get_context from keycardai.mcp.server.auth import ( ApplicationCredential, ClientSecret, @@ -60,6 +65,7 @@ "Context", "DefaultClientFactory", "EKSWorkloadIdentity", + "GrantDependency", "JWTVerifier", "MissingContextError", "NoneAuth", @@ -73,6 +79,7 @@ "get_claims", "get_token_debug_info", "introspect", + "override_access_context", ] logger = logging.getLogger(__name__) @@ -176,6 +183,143 @@ def get_token_debug_info(access_token: str) -> dict[str, Any]: return {"error": "Failed to parse token"} +# FastMCP context state key under which the AccessContext is stored. +# Reading it via ctx.get_state(KEYCARD_STATE_KEY) is deprecated in favor of +# declaring an AccessContext parameter; the key remains written during the +# deprecation window and for AccessContext.from_context(). +KEYCARD_STATE_KEY = "keycardai" + +# Testing seam: when set, grant() skips token acquisition and exchange and +# yields this AccessContext instead. Set via override_access_context(). +_access_context_override: ContextVar[AccessContext | None] = ContextVar( + "keycardai_access_context_override", default=None +) + + +@contextmanager +def override_access_context(access_context: AccessContext): + """Force Keycard grants to resolve to the given AccessContext. + + Public testing seam: while the context manager is active, every grant + resolution (injected-parameter or decorator form) skips caller-token + lookup and RFC 8693 exchange and produces ``access_context`` instead. + This is the supported way to fake delegated access in tests without + patching module internals. + + Args: + access_context: The AccessContext instance to inject into tools. + + Example: + ```python + from keycardai.fastmcp import AccessContext, override_access_context + from keycardai.oauth.types.models import TokenResponse + + access = AccessContext() + access.set_token("https://api.example.com", TokenResponse( + access_token="fake", token_type="Bearer", + )) + with override_access_context(access): + result = await my_tool.run(...) + ``` + """ + token = _access_context_override.set(access_context) + try: + yield access_context + finally: + _access_context_override.reset(token) + + +def _annotation_matches(annotation: Any, target: type) -> bool: + """Check whether a parameter annotation refers to ``target``. + + Handles resolved classes, subclasses, ``X | None`` / Optional unions, + ``Annotated[X, ...]``, and unresolved string annotations (as produced by + ``from __future__ import annotations`` when get_type_hints cannot resolve + them). + """ + if annotation is inspect.Parameter.empty or annotation is None: + return False + if annotation is target: + return True + if inspect.isclass(annotation) and issubclass(annotation, target): + return True + if isinstance(annotation, str): + # Unresolvable forward reference: match on the bare class name. + # Reached only when get_type_hints failed for the whole function, so a + # same-named unrelated class can false-positive here; the tradeoff is + # accepted to keep TYPE_CHECKING-only imports working. + parts = [part.strip() for part in annotation.split("|")] + return any(part.split(".")[-1] == target.__name__ for part in parts) + origin = get_origin(annotation) + if origin is Annotated: + return _annotation_matches(get_args(annotation)[0], target) + if origin is Union or origin is UnionType or isinstance(annotation, UnionType): + return any(_annotation_matches(arg, target) for arg in get_args(annotation)) + return False + + +def _find_param_of_type(func: Callable, target: type) -> str | None: + """Find the first parameter of ``func`` annotated with ``target``. + + Resolves string annotations via get_type_hints so functions defined in + modules using ``from __future__ import annotations`` are handled + correctly. + """ + signature = inspect.signature(func) + try: + hints = get_type_hints(func, include_extras=True) + except Exception: + hints = {} + for name, parameter in signature.parameters.items(): + annotation = hints.get(name, parameter.annotation) + if _annotation_matches(annotation, target): + return name + return None + + +def _get_context(*args, **kwargs) -> Context | None: + """Find a FastMCP Context instance in a call's arguments.""" + for value in args: + if isinstance(value, Context): + return value + for value in kwargs.values(): + if isinstance(value, Context): + return value + return None + + +def _current_context_or_none() -> Context | None: + """Return the active FastMCP request context, or None outside a request.""" + try: + return get_context() + except RuntimeError: + return None + + +async def _call_func(is_async_func: bool, func: Callable, *args, **kwargs): + if is_async_func: + return await func(*args, **kwargs) + return func(*args, **kwargs) + + +def _scope_for( + request_scopes: str | list[str] | dict[str, str | list[str]] | None, + resource: str, +) -> str | None: + """Resolve the RFC 8693 scope string to request for a resource.""" + if request_scopes is None: + return None + value = ( + request_scopes.get(resource) + if isinstance(request_scopes, dict) + else request_scopes + ) + if value is None: + return None + scope = " ".join(value) if isinstance(value, list) else value + return scope or None + + class AccessContext: """Context object that provides access to exchanged tokens for specific resources. @@ -292,6 +436,172 @@ def access(self, resource: str) -> TokenResponse: return self._access_tokens[resource] + @classmethod + async def from_context(cls, ctx: Context) -> AccessContext: + """Read the AccessContext stored on a FastMCP Context. + + Escape hatch for helpers called from inside tools that receive the + FastMCP ``Context`` but not the injected ``AccessContext`` parameter. + Prefer declaring the parameter directly on the tool: + ``access: AccessContext = auth_provider.grant(...)``. + + Args: + ctx: The FastMCP request context. + + Returns: + The AccessContext written by a grant for this request. If no + grant ran, returns an AccessContext with a global error recorded + (this method never raises). + """ + state = await ctx.get_state(KEYCARD_STATE_KEY) + if isinstance(state, AccessContext): + return state + access_context = cls() + access_context.set_error({ + "message": ( + "No Keycard access context is available on this request. " + "Declare an AccessContext parameter with auth_provider.grant(...) " + "or apply the grant decorator to the tool." + ), + }) + return access_context + + +class GrantDependency(Dependency[AccessContext]): + """Injectable dependency that performs delegated token exchange. + + Returned by :meth:`AuthProvider.grant`. One object supports both idioms: + + - **Injected parameter (preferred)**: used as a parameter default, FastMCP + resolves it per request via ``fastmcp.dependencies`` and injects the + populated :class:`AccessContext`. The parameter is excluded from the + tool's input schema. + + ```python + @mcp.tool() + async def get_github_user( + access: AccessContext = auth_provider.grant("https://api.github.com"), + ) -> dict: + token = access.access("https://api.github.com").access_token + ... + ``` + + - **Decorator (deprecated result access via get_state)**: applied with + ``@auth_provider.grant(...)``, the same object wraps the function. If + the function declares an :class:`AccessContext` parameter it is + injected there; otherwise a :class:`DeprecationWarning` is emitted at + decoration time and the result must be read via + ``await ctx.get_state("keycardai")``. + + Errors are recorded on the returned AccessContext, never raised + (see :meth:`AccessContext.get_errors`). Multi-resource grants are + all-or-nothing: if any exchange fails, the AccessContext carries the + failing resource's error and no tokens, including tokens for resources + that exchanged successfully before the failure. + + Instances are stateless between calls: all per-request state lives on the + AccessContext produced by each resolution, so a single instance is safe to + share across concurrent requests. + """ + + def __init__( + self, + provider: AuthProvider, + resources: str | list[str], + request_scopes: str | list[str] | dict[str, str | list[str]] | None = None, + ): + self._provider = provider + self._resources = [resources] if isinstance(resources, str) else list(resources) + self._request_scopes = request_scopes + + async def __aenter__(self) -> AccessContext: + access_context = await self._provider._build_access_context( + self._resources, self._request_scopes + ) + # Dual-write to FastMCP context state while ctx.get_state("keycardai") + # remains supported; also backs AccessContext.from_context(). + ctx = _current_context_or_none() + if ctx is not None: + await ctx.set_state(KEYCARD_STATE_KEY, access_context, serializable=False) + return access_context + + def __call__(self, func: Callable) -> Callable: + """Apply as a decorator, preserving the ``@auth_provider.grant(...)`` spelling. + + The decorated function must declare an :class:`AccessContext` + parameter (injected, hidden from the tool schema) or a FastMCP + ``Context`` parameter (deprecated: result read via + ``ctx.get_state("keycardai")``). + + Raises: + MissingContextError: If the function declares neither an + AccessContext parameter nor a Context parameter, or if no + Context can be found at call time on the deprecated path. + """ + provider = self._provider + resources = self._resources + request_scopes = self._request_scopes + + access_param = _find_param_of_type(func, AccessContext) + ctx_param = _find_param_of_type(func, Context) + if access_param is None and ctx_param is None: + raise MissingContextError( + function_name=func.__name__, + parameters=list(inspect.signature(func).parameters.keys()) + ) + if access_param is None: + warnings.warn( + f"Tool '{func.__name__}' uses the grant decorator without declaring " + "an AccessContext parameter; reading the result via " + 'ctx.get_state("keycardai") is deprecated. Declare a parameter ' + "like `access: AccessContext = auth_provider.grant(...)` instead.", + DeprecationWarning, + stacklevel=2, + ) + + is_async_func = inspect.iscoroutinefunction(func) + + @wraps(func) + async def wrapper(*args, **kwargs) -> Any: + _ctx = _get_context(*args, **kwargs) or _current_context_or_none() + if _ctx is None and access_param is None: + raise MissingContextError( + function_name=func.__name__, + parameters=[type(arg).__name__ for arg in args] + list(kwargs.keys()), + runtime_context=True + ) + + _access_context = await provider._build_access_context( + resources, request_scopes + ) + if _ctx is not None: + # Dual-write during the get_state deprecation window. + await _ctx.set_state(KEYCARD_STATE_KEY, _access_context, serializable=False) + if access_param is not None: + kwargs[access_param] = _access_context + + logger.debug(f"Executing decorated function: {func.__name__}") + return await _call_func(is_async_func, func, *args, **kwargs) + + if access_param is not None: + # Hide the injected AccessContext parameter from the wrapper's + # public signature so FastMCP excludes it from the tool schema + # and never treats it as a user-supplied argument. + signature = inspect.signature(func) + wrapper.__signature__ = signature.replace( + parameters=[ + parameter + for name, parameter in signature.parameters.items() + if name != access_param + ] + ) + wrapper.__annotations__ = { + name: annotation + for name, annotation in wrapper.__annotations__.items() + if name != access_param + } + return wrapper + class AuthProvider: """Keycard authentication provider for FastMCP. @@ -303,7 +613,7 @@ class AuthProvider: Example: ```python - from fastmcp import FastMCP, Context + from fastmcp import FastMCP from keycardai.fastmcp import AuthProvider, AccessContext # Using zone_id (recommended) @@ -337,18 +647,19 @@ class AuthProvider: auth = auth_provider.get_remote_auth_provider() mcp = FastMCP("My Protected Service", auth=auth) - # Use grant decorator for token exchange + # Declare a grant as a typed tool parameter for token exchange @mcp.tool() - @auth_provider.grant("https://api.example.com") - async def my_tool(ctx: Context, user_id: str): - # Use access context to check the status of the token exchange - # and handle the error state accordingly - access_context: AccessContext = await ctx.get_state("keycardai") - if access_context.has_errors(): + async def my_tool( + user_id: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ): + # Use the injected access context to check the status of the + # token exchange and handle the error state accordingly + if access.has_errors(): print("Failed to obtain access token for resource") - print(f"Error: {access_context.get_errors()}") + print(f"Error: {access.get_errors()}") return - token = access_context.access("https://api.example.com").access_token + token = access.access("https://api.example.com").access_token # Use token to call external API return f"Data for user {user_id}" ``` @@ -608,16 +919,26 @@ def grant( resources: str | list[str], *, request_scopes: str | list[str] | dict[str, str | list[str]] | None = None, - ): - """Decorator for automatic delegated token exchange. - - This decorator automates the OAuth token exchange process for accessing - external resources on behalf of authenticated users. It follows the FastMCP - Context namespace pattern, making tokens available through ctx.get_state("keycardai"). - - The returned value is an instance of AccessContext, which can be used to check the status of the token exchange - - The decorator avoids raising exceptions, and instead sets the error state in the AccessContext. + # Declared Any (the FastAPI Depends() convention) so both supported + # idioms type-check: as a parameter default the checker takes the type + # from the `access: AccessContext` annotation, and as a decorator the + # runtime GrantDependency.__call__ applies. A precise return type + # breaks one idiom or the other. Narrow to AccessContext when the + # decorator path is removed. + ) -> Any: + """Delegated token exchange for one or more resources. + + Returns a :class:`GrantDependency` (typed as ``Any`` so both call + idioms type-check) that automates the OAuth token exchange process + (RFC 8693) for accessing external resources on behalf of + authenticated users. Use it as a typed parameter default (preferred) + or as a decorator (the parameter-less get_state access is deprecated). + + The injected value is an instance of AccessContext, which can be used + to check the status of the token exchange. + + Grant resolution avoids raising exceptions, and instead sets the error + state in the AccessContext. Args: resources: Target resource URL(s) for token exchange. @@ -642,7 +963,7 @@ def grant( Usage: ```python - from fastmcp import FastMCP, Context + from fastmcp import FastMCP from keycardai.fastmcp import AuthProvider, AccessContext auth_provider = AuthProvider(zone_id="abc1234", mcp_base_url="http://localhost:8000") @@ -650,199 +971,158 @@ def grant( mcp = FastMCP("Server", auth=auth) @mcp.tool() - @auth_provider.grant("https://api.example.com") - async def my_tool(ctx: Context, user_id: str): - # Access token available through context namespace - access_context: AccessContext = await ctx.get_state("keycardai") - if access_context.has_errors(): + async def my_tool( + user_id: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ): + if access.has_errors(): print("Failed to obtain access token for resource") - print(f"Error: {access_context.get_errors()}") + print(f"Error: {access.get_errors()}") return - token = access_context.access("https://api.example.com").access_token + token = access.access("https://api.example.com").access_token headers = {"Authorization": f"Bearer {token}"} # Use headers to call external API return f"Data for {user_id}" # Request a scope for a single resource @mcp.tool() - @auth_provider.grant( - "https://api.example.com", - request_scopes="read", - ) - async def scoped_tool(ctx: Context): + async def scoped_tool( + access: AccessContext = auth_provider.grant( + "https://api.example.com", + request_scopes="read", + ), + ): ... # Per-resource scopes when exchanging for multiple resources @mcp.tool() - @auth_provider.grant( - ["https://api1.example.com", "https://api2.example.com"], - request_scopes={ - "https://api1.example.com": "read", - "https://api2.example.com": ["read", "write"], - }, - ) - async def multi_tool(ctx: Context): + async def multi_tool( + access: AccessContext = auth_provider.grant( + ["https://api1.example.com", "https://api2.example.com"], + request_scopes={ + "https://api1.example.com": "read", + "https://api2.example.com": ["read", "write"], + }, + ), + ): ... ``` - The decorated function must: - - Have a Context parameter from FastMCP (e.g., `ctx: Context`) - - Be an async function (required for `await ctx.get_state()`) + The decorator form remains supported from the same object: + ``@auth_provider.grant(...)`` above the function. On that path the + function must declare an AccessContext parameter (injected, hidden + from the tool schema) or a FastMCP Context parameter; without an + AccessContext parameter a DeprecationWarning is emitted and the + result must be read via ``await ctx.get_state("keycardai")``. Raises: - MissingContextError: If the decorated function doesn't have a Context parameter - or if Context cannot be found in function arguments at runtime + MissingContextError: Decorator form only: if the decorated function + declares neither an AccessContext parameter nor a + Context parameter, or if Context cannot be found + at call time on the deprecated get_state path. Error handling: - - Returns structured error response if token exchange fails + - Records structured errors on the AccessContext if token exchange fails - Preserves original function signature and behavior - Provides detailed error messages for debugging """ - def _has_context(func: Callable) -> bool: - sig = inspect.signature(func) - for value in sig.parameters.values(): - if value.annotation == Context: - return True - return False - - def _get_context(*args, **kwargs) -> Context | None: - for value in args: - if isinstance(value, Context): - return value - for value in kwargs.values(): - if isinstance(value, Context): - return value - return None - - def _scope_for(resource: str) -> str | None: - """Resolve the RFC 8693 scope string to request for a resource.""" - if request_scopes is None: - return None - value = ( - request_scopes.get(resource) - if isinstance(request_scopes, dict) - else request_scopes - ) - if value is None: - return None - scope = " ".join(value) if isinstance(value, list) else value - return scope or None - - async def _set_error(error: dict[str, str], resource: str | None, access_context: AccessContext, ctx: Context): - """Helper to set error context and call function.""" - if resource: - access_context.set_resource_error(resource, error) - else: - access_context.set_error(error) - await ctx.set_state("keycardai", access_context, serializable=False) - - async def _call_func(is_async_func: bool, func: Callable, *args, **kwargs): - if is_async_func: - return await func(*args, **kwargs) - else: - return func(*args, **kwargs) + return GrantDependency(self, resources, request_scopes) - def decorator(func: Callable) -> Callable: - is_async_func = inspect.iscoroutinefunction(func) - if not _has_context(func): - raise MissingContextError( - function_name=func.__name__, - parameters=list(inspect.signature(func).parameters.keys()) - ) + async def _build_access_context( + self, + resources: list[str], + request_scopes: str | list[str] | dict[str, str | list[str]] | None = None, + ) -> AccessContext: + """Acquire the caller token and exchange it for each resource. + + Never raises: failures are recorded on the returned AccessContext as + a global error (token acquisition) or a resource error (exchange). + Multi-resource grants are all-or-nothing: the first failed exchange + stops the loop and no tokens are populated, including ones already + exchanged successfully. + Honors the override_access_context() testing seam. + """ + override = _access_context_override.get() + if override is not None: + return override - @wraps(func) - async def wrapper(*args, **kwargs) -> Any: - _ctx = _get_context(*args, **kwargs) - if _ctx is None: - raise MissingContextError( - function_name=func.__name__, - parameters=[type(arg).__name__ for arg in args] + list(kwargs.keys()), - runtime_context=True + logger.debug(f"Starting token exchange for resources: {resources}") + _access_context = AccessContext() + try: + _user_token = get_access_token() + if not _user_token: + logger.warning("No authentication token available") + _access_context.set_error({ + "message": "No authentication token available. Please ensure you're properly authenticated.", + }) + return _access_context + logger.introspect(f"User token retrieved: {get_token_debug_info(_user_token.token)}") + except Exception as e: + logger.error("Failed to get access token") + _access_context.set_error({ + "message": "Failed to get access token from the request context. Please ensure you're properly authenticated.", + "raw_error": str(e), + }) + return _access_context + + _access_tokens = {} + for resource in resources: + logger.debug(f"Exchanging token for resource: {resource}") + try: + if self.application_credential: + logger.debug(f"Using application credential: {type(self.application_credential).__name__}") + # auth_info context is used by application credential implementation + # to prepare correct assertions in the token exchange request. + # For WebIdentity, use the stable WIF key_id so the client assertion + # JWT has a predictable `iss` that can be pre-registered in Keycard. + # Falling back to the DCR client_id would produce an ephemeral `ua:...` + # identifier that changes on every restart and cannot be pre-registered. + _resource_client_id = ( + self.application_credential.identity_manager.key_id + if hasattr(self.application_credential, "identity_manager") + else self.client.config.client_id or "" + ) + _auth_info = { + "resource_client_id": _resource_client_id, + "resource_server_url": self.mcp_base_url, + "zone_id": "", + } + _token_exchange_request = await self.application_credential.prepare_token_exchange_request( + client=self.client, + subject_token=_user_token.token, + resource=resource, + auth_info=_auth_info, + ) + else: + _token_exchange_request = TokenExchangeRequest( + subject_token=_user_token.token, + resource=resource, + subject_token_type="urn:ietf:params:oauth:token-type:access_token", ) - _resource_list = [resources] if isinstance(resources, str) else resources - logger.debug(f"Starting token exchange for resources: {_resource_list}") - - _access_context = AccessContext() - try: - _user_token = get_access_token() - if not _user_token: - logger.warning(f"No authentication token available for {func.__name__}") - await _set_error({ - "message": "No authentication token available. Please ensure you're properly authenticated.", - }, None, _access_context, _ctx) - return await _call_func(is_async_func, func, *args, **kwargs) - logger.introspect(f"User token retrieved: {get_token_debug_info(_user_token.token)}") - except Exception as e: - logger.error("Failed to get access token") - await _set_error({ - "message": "Failed to get access token from context. Ensure the Context parameter is properly annotated.", - "raw_error": str(e), - }, None, _access_context, _ctx) - return await _call_func(is_async_func, func, *args, **kwargs) - - _access_tokens = {} - for resource in _resource_list: - logger.debug(f"Exchanging token for resource: {resource}") - try: - if self.application_credential: - logger.debug(f"Using application credential: {type(self.application_credential).__name__}") - # auth_info context is used by application credential implementation - # to prepare correct assertions in the token exchange request. - # For WebIdentity, use the stable WIF key_id so the client assertion - # JWT has a predictable `iss` that can be pre-registered in Keycard. - # Falling back to the DCR client_id would produce an ephemeral `ua:...` - # identifier that changes on every restart and cannot be pre-registered. - _resource_client_id = ( - self.application_credential.identity_manager.key_id - if hasattr(self.application_credential, "identity_manager") - else self.client.config.client_id or "" - ) - _auth_info = { - "resource_client_id": _resource_client_id, - "resource_server_url": self.mcp_base_url, - "zone_id": "", - } - _token_exchange_request = await self.application_credential.prepare_token_exchange_request( - client=self.client, - subject_token=_user_token.token, - resource=resource, - auth_info=_auth_info, - ) - else: - _token_exchange_request = TokenExchangeRequest( - subject_token=_user_token.token, - resource=resource, - subject_token_type="urn:ietf:params:oauth:token-type:access_token", - ) - - _scope = _scope_for(resource) - if _scope: - _token_exchange_request.scope = _scope - - _token_response = await self.client.exchange_token(_token_exchange_request) - - _access_tokens[resource] = _token_response - logger.debug(f"Token exchange successful for {resource}") - logger.introspect(f"Token details for {resource}: {get_token_debug_info(_token_response.access_token)}") - except Exception as e: - logger.error(f"Token exchange failed for {resource}") - _error_dict: dict[str, str] = { - "message": f"Token exchange failed for {resource}", - } - if hasattr(e, "error"): - _error_dict["code"] = e.error - if hasattr(e, "error_description") and e.error_description: - _error_dict["description"] = e.error_description - if not hasattr(e, "error"): - _error_dict["raw_error"] = str(e) - await _set_error(_error_dict, resource, _access_context, _ctx) - return await _call_func(is_async_func, func, *args, **kwargs) - - logger.debug(f"All token exchanges completed. Setting access context with {len(_access_tokens)} token(s)") - _access_context.set_bulk_tokens(_access_tokens) - await _ctx.set_state("keycardai", _access_context, serializable=False) - logger.debug(f"Executing decorated function: {func.__name__}") - return await _call_func(is_async_func, func, *args, **kwargs) - return wrapper - return decorator + _scope = _scope_for(request_scopes, resource) + if _scope: + _token_exchange_request.scope = _scope + + _token_response = await self.client.exchange_token(_token_exchange_request) + + _access_tokens[resource] = _token_response + logger.debug(f"Token exchange successful for {resource}") + logger.introspect(f"Token details for {resource}: {get_token_debug_info(_token_response.access_token)}") + except Exception as e: + logger.error(f"Token exchange failed for {resource}") + _error_dict: dict[str, str] = { + "message": f"Token exchange failed for {resource}", + } + if hasattr(e, "error"): + _error_dict["code"] = e.error + if hasattr(e, "error_description") and e.error_description: + _error_dict["description"] = e.error_description + if not hasattr(e, "error"): + _error_dict["raw_error"] = str(e) + _access_context.set_resource_error(resource, _error_dict) + return _access_context + + logger.debug(f"All token exchanges completed. Populating access context with {len(_access_tokens)} token(s)") + _access_context.set_bulk_tokens(_access_tokens) + return _access_context diff --git a/packages/fastmcp/src/keycardai/fastmcp/testing/__init__.py b/packages/fastmcp/src/keycardai/fastmcp/testing/__init__.py index 06af6099..f735c807 100644 --- a/packages/fastmcp/src/keycardai/fastmcp/testing/__init__.py +++ b/packages/fastmcp/src/keycardai/fastmcp/testing/__init__.py @@ -5,9 +5,13 @@ Components: - mock_access_context: Context manager for mocking authentication in tests +- override_access_context: Lower-level seam that injects a caller-built AccessContext + +Both utilities work for tools using the injected-parameter form +(``access: AccessContext = auth_provider.grant(...)``) and the decorator form. Example: - from keycardai.mcp.integrations.fastmcp.testing import mock_access_context + from keycardai.fastmcp.testing import mock_access_context # Test successful authentication with default token with mock_access_context(): @@ -27,10 +31,24 @@ # Test error scenarios with mock_access_context(has_errors=True, error_message="Auth failed"): # Your test code here - access_context.has_errors() will return True + + # Full control: build the AccessContext yourself + from keycardai.fastmcp import AccessContext + from keycardai.fastmcp.testing import override_access_context + from keycardai.oauth.types.models import TokenResponse + + access = AccessContext() + access.set_token("https://api.example.com", TokenResponse( + access_token="fake", token_type="Bearer", + )) + with override_access_context(access): + # Your test code here """ +from ..provider import override_access_context from .test_utils import mock_access_context __all__ = [ "mock_access_context", + "override_access_context", ] diff --git a/packages/fastmcp/src/keycardai/fastmcp/testing/test_utils.py b/packages/fastmcp/src/keycardai/fastmcp/testing/test_utils.py index e50e9a61..501921ea 100644 --- a/packages/fastmcp/src/keycardai/fastmcp/testing/test_utils.py +++ b/packages/fastmcp/src/keycardai/fastmcp/testing/test_utils.py @@ -1,8 +1,46 @@ from contextlib import contextmanager -from unittest.mock import Mock, patch from keycardai.oauth.types.models import TokenResponse +from ..provider import AccessContext, override_access_context + + +class _MockAccessContext(AccessContext): + """AccessContext preloaded for tests. + + Behaves like a real AccessContext, with one addition: when constructed + with a ``default_token``, :meth:`access` returns that token for any + resource that has no explicit token or error. When a resource lookup + fails, the miss is recorded as a resource error so ``has_errors()`` + reflects it, matching how grant resolution records failures. + """ + + def __init__(self, default_token: str | None = None): + super().__init__() + self._default_token = default_token + + def access(self, resource: str) -> TokenResponse: + if ( + self._default_token is not None + and not self.has_errors() + and resource not in self._access_tokens + ): + return TokenResponse( + access_token=self._default_token, + token_type="Bearer", + ) + try: + return super().access(resource) + except Exception: + # Recording the miss on a read path is test-double behavior only: + # it keeps has_errors() observable after a failed lookup, which + # tests written against the original mock rely on. + if not self.has_error() and not self.has_resource_error(resource): + self.set_resource_error( + resource, {"message": f"Resource not granted: {resource}"} + ) + raise + @contextmanager def mock_access_context( @@ -13,12 +51,25 @@ def mock_access_context( ): """Mock the authentication system for testing. + Builds an :class:`AccessContext` from the given tokens or error state and + installs it through the public :func:`override_access_context` seam, so + every grant resolution (injected-parameter or decorator form) yields it + without touching real token acquisition or exchange. No module internals + are patched. + Args: access_token: Default access token to return for any resource (str) resource_tokens: Dict mapping resource URLs to specific access tokens (dict[str, str]) has_errors: Whether the access context should report errors (bool) error_message: Error message to return when has_errors=True (str) + Note: + The bare ``access_token`` form answers for **any** resource, including + ones the tool never granted, so it cannot catch a mistyped resource URL + in an ``access(...)`` call; in production that raises + :class:`ResourceAccessError`. Use ``resource_tokens={...}`` when the + test should enforce which resources the tool reads. + Examples: # 1. Default - always returns access token with mock_access_context(): @@ -34,65 +85,25 @@ def mock_access_context( "https://api.other.com": "token_456" }): # Will return specific tokens for each resource - # Any resource not in the dict will set has_errors=True with "Resource not granted" message + # Any resource not in the dict raises ResourceAccessError and + # records a "Resource not granted" error on the context # 4. Returns error set to true and error message with mock_access_context(has_errors=True, error_message="Auth failed"): # Will report errors with the specified message """ - with patch('keycardai.fastmcp.provider.AccessContext') as mock_access_context_class, \ - patch('keycardai.fastmcp.provider.get_access_token') as mock_get_access_token: - - mock_access_context_instance = Mock() - mock_access_context_instance.has_errors.return_value = has_errors - - if has_errors: - # Return proper error structure matching AccessContext.get_errors() - mock_access_context_instance.get_errors.return_value = { - "resources": {}, - "error": {"message": error_message} - } - mock_access_context_instance.access.side_effect = Exception(error_message) - else: - def mock_access_method(resource_url): - if resource_tokens is not None: - if resource_url in resource_tokens: - # Return proper TokenResponse object - return TokenResponse( - access_token=resource_tokens[resource_url], - token_type="Bearer" - ) - else: - # Resource not granted - set error state and raise exception - mock_access_context_instance.has_errors.return_value = True - mock_access_context_instance.get_errors.return_value = { - "resources": { - resource_url: {"message": f"Resource not granted: {resource_url}"} - }, - "error": None - } - from keycardai.mcp.server.exceptions import ResourceAccessError - raise ResourceAccessError() - else: - # Return proper TokenResponse object - return TokenResponse( - access_token=access_token, - token_type="Bearer" - ) - - mock_access_context_instance.access = mock_access_method - mock_access_context_instance.get_errors.return_value = { - "resources": {}, - "error": None - } - - mock_access_context_instance.set_bulk_tokens = Mock() - mock_access_context_instance.set_error = Mock() - mock_access_context_instance.set_resource_error = Mock() - mock_access_context_class.return_value = mock_access_context_instance - - mock_user_token = Mock() - mock_user_token.token = "user_jwt_token" - mock_get_access_token.return_value = mock_user_token - - yield mock_access_context_instance + access_context = _MockAccessContext( + default_token=None if resource_tokens is not None else access_token + ) + + if has_errors: + access_context.set_error({"message": error_message}) + elif resource_tokens is not None: + for resource, token in resource_tokens.items(): + access_context.set_token( + resource, + TokenResponse(access_token=token, token_type="Bearer"), + ) + + with override_access_context(access_context): + yield access_context diff --git a/packages/fastmcp/tests/integration/test_grant_dependency.py b/packages/fastmcp/tests/integration/test_grant_dependency.py new file mode 100644 index 00000000..f4012ca9 --- /dev/null +++ b/packages/fastmcp/tests/integration/test_grant_dependency.py @@ -0,0 +1,369 @@ +"""Integration tests for grant-as-dependency typed injection. + +Covers the GrantDependency returned by AuthProvider.grant(): +- injected-parameter path resolved through FastMCP's dependency machinery +- decorator path (still supported from the same object), including the + decoration-time DeprecationWarning and dual-write to context state +- error-capture contract (exchange failures recorded, never raised) +- AccessContext.from_context() escape hatch +- override_access_context() public testing seam +""" + +import inspect +import warnings +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from fastmcp import Context +from fastmcp.dependencies import Dependency +from fastmcp.server.dependencies import ( + AccessToken, + without_injected_parameters, +) + +from keycardai.fastmcp import ( + AccessContext, + AuthProvider, + GrantDependency, +) +from keycardai.fastmcp.testing import override_access_context +from keycardai.mcp.server.exceptions import MissingContextError +from keycardai.oauth.types.models import TokenResponse + + +def create_mock_context(): + """Create a mock Context with real state management.""" + mock_context = Mock(spec=Context) + context_state = {} + + async def mock_set_state(key: str, value, *, serializable=True): + context_state[key] = value + + async def mock_get_state(key: str): + return context_state.get(key) + + mock_context.set_state = mock_set_state + mock_context.get_state = mock_get_state + return mock_context + + +@pytest.fixture +def auth_provider(auth_provider_config, mock_client_factory): + return AuthProvider(**auth_provider_config, client_factory=mock_client_factory) + + +class TestGrantDependencyInjection: + """Injected-parameter path: access: AccessContext = auth_provider.grant(...).""" + + def test_grant_returns_fastmcp_dependency(self, auth_provider): + dep = auth_provider.grant("https://api.example.com") + assert isinstance(dep, GrantDependency) + assert isinstance(dep, Dependency) + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_injected_param_happy_path(self, mock_get_token, auth_provider): + """FastMCP's dependency machinery injects a populated AccessContext.""" + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + + async def get_data( + user_id: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ) -> str: + assert not access.has_errors() + token = access.access("https://api.example.com").access_token + return f"{user_id}:{token}" + + # Resolve through FastMCP's own dependency machinery. + wrapper = without_injected_parameters(get_data) + result = await wrapper(user_id="user123") + assert result == "user123:exchanged_token_123" + + def test_injected_param_hidden_from_signature(self, auth_provider): + """The AccessContext parameter is excluded from the tool's public signature.""" + + async def get_data( + user_id: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ) -> str: + return user_id + + wrapper = without_injected_parameters(get_data) + assert list(inspect.signature(wrapper).parameters.keys()) == ["user_id"] + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_injected_param_multiple_resources(self, mock_get_token, auth_provider): + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + dep = auth_provider.grant( + ["https://api1.example.com", "https://api2.example.com"] + ) + + async with dep as access: + assert access.access("https://api1.example.com").access_token == "token_api1_123" + assert access.access("https://api2.example.com").access_token == "token_api2_456" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_exchange_failure_recorded_not_raised(self, mock_get_token, auth_provider): + """Error-capture contract: __aenter__ never raises on exchange failure.""" + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + failing_client = AsyncMock() + failing_client.exchange_token.side_effect = Exception("Exchange failed") + auth_provider.client = failing_client + + dep = auth_provider.grant("https://api.example.com") + async with dep as access: + assert access.has_errors() + assert access.has_resource_error("https://api.example.com") + error = access.get_resource_error("https://api.example.com") + assert error["message"] == "Token exchange failed for https://api.example.com" + assert error["raw_error"] == "Exchange failed" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_missing_caller_token_recorded_not_raised(self, mock_get_token, auth_provider): + """Error-capture contract: missing caller token becomes a global error.""" + mock_get_token.return_value = None + + dep = auth_provider.grant("https://api.example.com") + async with dep as access: + assert access.has_error() + assert "No authentication token available" in access.get_error()["message"] + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_request_scopes_forwarded(self, mock_get_token, auth_provider): + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + captured = {} + + async def capturing_exchange(request): + captured[request.resource] = request + return TokenResponse(access_token="exchanged", token_type="Bearer") + + capturing_client = AsyncMock() + capturing_client.exchange_token.side_effect = capturing_exchange + auth_provider.client = capturing_client + + dep = auth_provider.grant("https://api.example.com", request_scopes="read") + async with dep as access: + assert not access.has_errors() + assert captured["https://api.example.com"].scope == "read" + + +class TestGrantDecoratorPath: + """Decorator spelling still works from the same GrantDependency object.""" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_decorator_still_works_and_warns_once(self, mock_get_token, auth_provider): + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + + @auth_provider.grant("https://api.example.com") + async def legacy_tool(ctx: Context, user_id: str): + access = await ctx.get_state("keycardai") + return access.access("https://api.example.com").access_token + + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + assert len(deprecations) == 1, "decoration emits exactly one DeprecationWarning per tool" + assert "legacy_tool" in str(deprecations[0].message) + assert "AccessContext" in str(deprecations[0].message) + + result = await legacy_tool(create_mock_context(), "user123") + assert result == "exchanged_token_123" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_decorator_with_access_param_no_warning(self, mock_get_token, auth_provider): + """Declaring an AccessContext parameter silences the deprecation warning.""" + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + + @auth_provider.grant("https://api.example.com") + async def typed_tool(user_id: str, access: AccessContext = None): + return access.access("https://api.example.com").access_token + + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + assert deprecations == [] + + # The injected parameter is hidden from the wrapper's public signature. + assert list(inspect.signature(typed_tool).parameters.keys()) == ["user_id"] + + result = await typed_tool("user123") + assert result == "exchanged_token_123" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_decorator_dual_writes_context_state(self, mock_get_token, auth_provider): + """Decorator keeps writing set_state("keycardai") during deprecation.""" + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + + @auth_provider.grant("https://api.example.com") + async def legacy_tool(ctx: Context): + return "ok" + + mock_context = create_mock_context() + await legacy_tool(mock_context) + + state = await mock_context.get_state("keycardai") + assert isinstance(state, AccessContext) + assert state.access("https://api.example.com").access_token == "exchanged_token_123" + + def test_decorator_without_context_or_access_param_raises(self, auth_provider): + with pytest.raises(MissingContextError): + + @auth_provider.grant("https://api.example.com") + def no_injection(user_id: str) -> str: + return user_id + + def test_decorator_handles_string_and_union_annotations(self, auth_provider): + """Context detection works with string annotations and Context | None.""" + + @auth_provider.grant("https://api.example.com") + async def string_annotated(ctx: "Context"): + return "ok" + + @auth_provider.grant("https://api.example.com") + async def union_annotated(ctx: Context | None = None): + return "ok" + + # Decoration succeeded for both; MissingContextError was not raised. + assert callable(string_annotated) + assert callable(union_annotated) + + +class TestFromContext: + """AccessContext.from_context() escape hatch for helpers inside tools.""" + + @pytest.mark.asyncio + @patch("keycardai.fastmcp.provider.get_access_token") + async def test_from_context_returns_stored_access_context(self, mock_get_token, auth_provider): + mock_get_token.return_value = AccessToken( + token="test_token", client_id="test_client", scopes=["test_scope"] + ) + + async def helper(ctx: Context) -> str: + access = await AccessContext.from_context(ctx) + return access.access("https://api.example.com").access_token + + @auth_provider.grant("https://api.example.com") + async def tool(ctx: Context): + return await helper(ctx) + + result = await tool(create_mock_context()) + assert result == "exchanged_token_123" + + @pytest.mark.asyncio + async def test_from_context_without_grant_records_error(self): + access = await AccessContext.from_context(create_mock_context()) + assert isinstance(access, AccessContext) + assert access.has_error() + assert "No Keycard access context" in access.get_error()["message"] + + +class TestEndToEndWithFastMCPServer: + """Full round trip through a real FastMCP server and in-memory client.""" + + @pytest.mark.asyncio + async def test_injected_param_through_real_server(self, auth_provider): + from fastmcp import Client, FastMCP + + mcp = FastMCP("test-server") + + @mcp.tool() + async def get_data( + user_id: str, + access: AccessContext = auth_provider.grant("https://api.example.com"), + ) -> str: + token = access.access("https://api.example.com").access_token + return f"{user_id}:{token}" + + fake = AccessContext() + fake.set_token( + "https://api.example.com", + TokenResponse(access_token="e2e_token", token_type="Bearer"), + ) + + with override_access_context(fake): + async with Client(mcp) as client: + tools = await client.list_tools() + schema_props = tools[0].inputSchema.get("properties", {}) + assert list(schema_props.keys()) == ["user_id"], ( + "AccessContext parameter must not appear in the tool input schema" + ) + + result = await client.call_tool("get_data", {"user_id": "user123"}) + assert result.content[0].text == "user123:e2e_token" + + +class TestOverrideAccessContextSeam: + """override_access_context() bypasses token acquisition and exchange.""" + + @pytest.mark.asyncio + async def test_override_short_circuits_dependency_resolution(self, auth_provider): + # No get_access_token patching and no client mocking: the override + # must short-circuit before either is touched. + auth_provider.client = None + + fake = AccessContext() + fake.set_token( + "https://api.example.com", + TokenResponse(access_token="seam_token", token_type="Bearer"), + ) + + async def get_data( + access: AccessContext = auth_provider.grant("https://api.example.com"), + ) -> str: + return access.access("https://api.example.com").access_token + + wrapper = without_injected_parameters(get_data) + with override_access_context(fake): + assert await wrapper() == "seam_token" + + @pytest.mark.asyncio + async def test_override_applies_to_decorator_path(self, auth_provider): + auth_provider.client = None + + fake = AccessContext() + fake.set_token( + "https://api.example.com", + TokenResponse(access_token="seam_token", token_type="Bearer"), + ) + + @auth_provider.grant("https://api.example.com") + async def legacy_tool(ctx: Context): + access = await ctx.get_state("keycardai") + return access.access("https://api.example.com").access_token + + with override_access_context(fake): + assert await legacy_tool(create_mock_context()) == "seam_token" + + @pytest.mark.asyncio + async def test_override_resets_after_exit(self, auth_provider): + fake = AccessContext() + with override_access_context(fake): + pass + + with patch("keycardai.fastmcp.provider.get_access_token", return_value=None): + dep = auth_provider.grant("https://api.example.com") + async with dep as access: + assert access is not fake diff --git a/packages/fastmcp/tests/test_examples.py b/packages/fastmcp/tests/test_examples.py index 7805b121..3ac623fe 100644 --- a/packages/fastmcp/tests/test_examples.py +++ b/packages/fastmcp/tests/test_examples.py @@ -1,9 +1,9 @@ """Smoke tests for package examples. -Verifies that each example in packages/mcp-fastmcp/examples/ imports cleanly -and exposes the expected objects. The @grant decorator validates function -signatures at import time, so a successful import confirms the example -is compatible with the current SDK API. +Verifies that each example in packages/fastmcp/examples/ imports cleanly +and exposes the expected objects. Grant declarations (parameter defaults and +decorators) are constructed and registered with FastMCP at import time, so a +successful import confirms the example is compatible with the current SDK API. """ import importlib.util @@ -73,7 +73,8 @@ def test_delegated_access_example_loads(): assert hasattr(mod, "auth") assert hasattr(mod, "mcp") # FastMCP's @mcp.tool() wraps functions in FunctionTool objects (not callable), - # but their existence confirms @grant validated the signatures at import time. + # but their existence confirms the grant parameter declarations were + # accepted by FastMCP's dependency machinery at registration time. assert hasattr(mod, "get_github_user") assert hasattr(mod, "list_github_repos") assert hasattr(mod, "main") diff --git a/packages/fastmcp/tests/test_mock_access_context.py b/packages/fastmcp/tests/test_mock_access_context.py new file mode 100644 index 00000000..dffc9929 --- /dev/null +++ b/packages/fastmcp/tests/test_mock_access_context.py @@ -0,0 +1,90 @@ +"""Tests for the mock_access_context testing utility. + +mock_access_context installs a preloaded AccessContext through the public +override_access_context seam; nothing here patches module internals, and +tools exercise the same grant resolution code paths as production. +""" + +import pytest +from fastmcp.server.dependencies import without_injected_parameters + +from keycardai.fastmcp import ( + AccessContext, + AuthProvider, + ResourceAccessError, + mock_access_context, +) + + +@pytest.fixture +def auth_provider(auth_provider_config, mock_client_factory): + return AuthProvider(**auth_provider_config, client_factory=mock_client_factory) + + +@pytest.mark.asyncio +async def test_default_token_for_any_resource(auth_provider): + async def tool( + access: AccessContext = auth_provider.grant("https://api.example.com"), + ) -> str: + assert not access.has_errors() + return access.access("https://api.example.com").access_token + + wrapper = without_injected_parameters(tool) + with mock_access_context(): + assert await wrapper() == "test_access_token" + with mock_access_context(access_token="my_token"): + assert await wrapper() == "my_token" + + +@pytest.mark.asyncio +async def test_resource_specific_tokens(auth_provider): + async def tool( + access: AccessContext = auth_provider.grant( + ["https://api.example.com", "https://api.other.com"] + ), + ) -> tuple[str, str]: + return ( + access.access("https://api.example.com").access_token, + access.access("https://api.other.com").access_token, + ) + + wrapper = without_injected_parameters(tool) + with mock_access_context(resource_tokens={ + "https://api.example.com": "token_123", + "https://api.other.com": "token_456", + }): + assert await wrapper() == ("token_123", "token_456") + + +def test_resource_not_granted_raises_and_records(): + with mock_access_context(resource_tokens={ + "https://api.example.com": "token_123", + }) as access: + with pytest.raises(ResourceAccessError): + access.access("https://api.other.com") + assert access.has_errors() + errors = access.get_errors() + assert "https://api.other.com" in errors["resources"] + + +def test_error_state(): + with mock_access_context(has_errors=True, error_message="Auth failed") as access: + assert access.has_errors() + assert access.get_errors()["error"] == {"message": "Auth failed"} + with pytest.raises(Exception, match=""): + access.access("https://api.example.com") + + +def test_yields_real_access_context(): + with mock_access_context() as access: + assert isinstance(access, AccessContext) + + +@pytest.mark.asyncio +async def test_works_with_decorator_path(auth_provider): + @auth_provider.grant("https://api.example.com") + async def tool(access: AccessContext = None) -> str: + return access.access("https://api.example.com").access_token + + with mock_access_context(access_token="decorator_token"): + assert await tool() == "decorator_token" diff --git a/uv.lock b/uv.lock index c0f5550e..bb40f5d6 100644 --- a/uv.lock +++ b/uv.lock @@ -209,9 +209,9 @@ name = "aiologic" version = "0.16.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "sniffio" }, - { name = "typing-extensions" }, - { name = "wrapt" }, + { name = "sniffio", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "wrapt", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a8/13/50b91a3ea6b030d280d2654be97c48b6ed81753a50286ee43c646ba36d3c/aiologic-0.16.0.tar.gz", hash = "sha256:c267ccbd3ff417ec93e78d28d4d577ccca115d5797cdbd16785a551d9658858f", size = 225952, upload-time = "2025-11-27T23:48:41.195Z" } wheels = [ @@ -1056,8 +1056,8 @@ name = "culsans" version = "0.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "aiologic" }, - { name = "typing-extensions" }, + { name = "aiologic", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d9/e3/49afa1bc180e0d28008ec6bcdf82a4072d1c7a41032b5b759b60814ca4b0/culsans-0.11.0.tar.gz", hash = "sha256:0b43d0d05dce6106293d114c86e3fb4bfc63088cfe8ff08ed3fe36891447fe33", size = 107546, upload-time = "2025-12-31T23:15:38.196Z" } wheels = [ @@ -2203,7 +2203,7 @@ test = [ [package.metadata] requires-dist = [ - { name = "fastmcp", specifier = ">=3.0.0" }, + { name = "fastmcp", specifier = ">=3.1.0" }, { name = "httpx", specifier = ">=0.27.2" }, { name = "keycardai-mcp", editable = "packages/mcp" }, { name = "keycardai-oauth", editable = "packages/oauth" }, @@ -3330,11 +3330,11 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "flatbuffers" }, - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, - { name = "packaging" }, - { name = "protobuf" }, - { name = "sympy" }, + { name = "flatbuffers", marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "packaging", marker = "python_full_version < '3.11'" }, + { name = "protobuf", marker = "python_full_version < '3.11'" }, + { name = "sympy", marker = "python_full_version < '3.11'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/15/41/3253db975a90c3ce1d475e2a230773a21cd7998537f0657947df6fb79861/onnxruntime-1.24.3-cp311-cp311-macosx_14_0_arm64.whl", hash = "sha256:3e6456801c66b095c5cd68e690ca25db970ea5202bd0c5b84a2c3ef7731c5a3c", size = 17332766, upload-time = "2026-03-05T17:18:59.714Z" }, @@ -3374,10 +3374,10 @@ resolution-markers = [ "python_full_version >= '3.11' and python_full_version < '3.13'", ] dependencies = [ - { name = "flatbuffers" }, - { name = "numpy", version = "2.4.4", source = { registry = "https://pypi.org/simple" } }, - { name = "packaging" }, - { name = "protobuf" }, + { name = "flatbuffers", marker = "python_full_version >= '3.11'" }, + { name = "numpy", version = "2.4.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "packaging", marker = "python_full_version >= '3.11'" }, + { name = "protobuf", marker = "python_full_version >= '3.11'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/c1/00/dccf702195572df51a40784fc939304595a0ae3577537d3b5be79273151a/onnxruntime-1.25.1-cp311-cp311-macosx_14_0_arm64.whl", hash = "sha256:5cf58ec7601120bb4370f0b868f794d3e3626db7b1b1dba366c27874b224e9de", size = 17762805, upload-time = "2026-04-27T22:00:45.336Z" }, @@ -5348,7 +5348,7 @@ name = "sympy" version = "1.14.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "mpmath" }, + { name = "mpmath", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/83/d3/803453b36afefb7c2bb238361cd4ae6125a569b4db67cd9e79846ba2d68c/sympy-1.14.0.tar.gz", hash = "sha256:d3d3fe8df1e5a0b42f0e7bdf50541697dbe7d23746e894990c030e2b05e72517", size = 7793921, upload-time = "2025-04-27T18:05:01.611Z" } wheels = [