Bound FastMCP injection claims to whole-signature evidence

docs/mcp-registration-idioms.md

@@ -113,9 +113,45 @@ id nobody serves is worse than a measured gap.
 
 ### What the request context is, and is not
 
-The server injects its `Context` on the parameter's **annotation**, so this
-reader drops a parameter only when it is annotated `Context` — including
-`Context | None` and `Context[ServerSession, None]`, and not `list[Context]`.
+Context injection is a **whole-signature selection**, not a rule applied
+independently to every parameter. The supported static subset excludes only
+the first established direct or nullable Context parameter, including aliases,
+qualified imports, quoted annotations and plain local subclasses. It first
+checks every annotation, including return, `*args` and `**kwargs`, because an
+unresolved annotation elsewhere can prevent the framework from selecting any
+parameter. Subsequent Context parameters stay in the inventory.
+
+The SDK and standalone FastMCP do not have identical rules. The pinned
+[SDK 1.27.2 utility](https://github.com/modelcontextprotocol/python-sdk/blob/v1.27.2/src/mcp/server/fastmcp/utilities/context_injection.py)
+resolves all type hints, selects the first match and inspects immediate generic
+arguments; its installed implementation selects `holder` for `list[Context]`.
+[Standalone FastMCP 2.14.5](https://github.com/PrefectHQ/fastmcp/blob/21221b4ab128e8dd71b5d9637fa70a9139511380/src/fastmcp/utilities/types.py)
+does not treat a container's Context element as membership and has a different
+raw-annotation fallback. Parameterized Context can itself materialize a class
+in Pydantic. The reader therefore leaves generic/parameterized/nested Context,
+uncertain annotated variadics, Context returns, and unresolved signature
+annotations outside its supported injection subset. It does not import a
+target server or evaluate its annotations to settle them.
+
+`unresolved_context_signature` names this function-level limitation, including
+cases with no ordinary parameter to carry it. A retained parameter whose
+injection cannot be established also carries `unresolved_context_identity` in
+the tool's surface gaps. Generic arguments and `Annotated` metadata are not
+discarded when deciding whether the signature is understood; only literal
+metadata is supported. An arbitrary external import may re-export Context or
+a subclass, so its package name alone never proves caller ownership. Canonical
+framework classes, builtins, plain local classes and narrowly recognized
+Pydantic `BaseModel` ancestry have static identity evidence; decorators,
+metaclasses and unresolved bases do not.
+
+Tool discovery and the medium confidence ceiling are retained for all these
+limits. The output names unresolved parameters and coverage rather than
+silently removing them. This is a bounded common subset, not full injection
+fidelity across framework versions, mixed framework families, import re-exports
+or arbitrary framework submodules (the inherited provenance gap is tracked in
+[#601](https://github.com/ThreeMoonsLab/agents-shipgate/issues/601)). A committed MCP export remains the stronger
+route when that complete published surface is available.
+
 The conventional-name list this package's other Python adapters share holds
 `config`, `context` and `runtime`, which are ordinary user-supplied inputs to
 an MCP tool: dropping them by name published an empty schema for a

src/agents_shipgate/inputs/mcp_idioms.py

@@ -485,6 +485,9 @@ class RegistrationSite:
     #: represent — the two are told apart by ``returns``, and only the second
     #: is a gap.
     returns_json_type: str | None = None
+    #: Whole-signature injection could not be established. This also covers
+    #: annotated variadics/returns, which have no SignatureParameter row.
+    context_injection_unresolved: bool = False
     #: Whether *this site alone* is evidence that the repository is an MCP
     #: server, independently of whether its name could be read.
     #:
@@ -2366,6 +2369,9 @@ def _python_site(
         if wrapped
         else _python_tool_name(node, decorator)
     )
+    parameters, context_injection_unresolved = (
+        (None, False) if wrapped else _python_signature(node, module, index, module_path)
+    )
     return RegistrationSite(
         idiom="py_fastmcp_decorator",
         name=name,
@@ -2377,9 +2383,8 @@ def _python_site(
         # Withheld with the name: the schema comes from the same object, so a
         # signature published beside an unreadable name would be a parameter
         # list for a function the server may not have registered.
-        parameters=(
-            None if wrapped else _python_signature(node, module, index, module_path)
-        ),
+        parameters=parameters,
+        context_injection_unresolved=context_injection_unresolved,
         returns=None if wrapped else _python_annotation(node.returns),
         returns_json_type=(
             None
@@ -2461,16 +2466,45 @@ def _python_signature(
     module: _PythonModule,
     index: PythonServerIndex,
     module_path: str | None,
-) -> tuple[SignatureParameter, ...]:
+) -> tuple[tuple[SignatureParameter, ...], bool]:
+    arguments = node.args
+    all_arguments = [
+        *arguments.posonlyargs, *arguments.args,
+        *([arguments.vararg] if arguments.vararg else []),
+        *arguments.kwonlyargs,
+        *([arguments.kwarg] if arguments.kwarg else []),
+    ]
+    identities = {
+        argument.arg: _python_context_injection(
+            argument.annotation, module, index, module_path
+        ) for argument in all_arguments
+    }
+    returns_identity = _python_context_injection(node.returns, module, index, module_path)
+    # SDK get_type_hints resolves the entire signature before selecting a
+    # parameter. Standalone FastMCP has a different raw-annotation fallback;
+    # neither result may be inferred by discarding an unresolved annotation.
+    unresolved = "unresolved" in identities.values() or returns_identity != "caller_supplied"
+    # Annotated variadics can win the framework's first-match selection but
+    # are not schema properties. Keep that unsupported selection visible.
+    unresolved |= any(
+        argument is not None and identities[argument.arg] != "caller_supplied"
+        for argument in (arguments.vararg, arguments.kwarg)
+    )
+    selected = None if unresolved else next(
+        (name for name, identity in identities.items() if identity == "framework_injected"), None
+    )
+
     def _parameter(
         argument: ast.arg, default: ast.expr | None
     ) -> SignatureParameter:
         return SignatureParameter(
             name=argument.arg,
             annotation=_python_annotation(argument.annotation),
             required=default is None,
-            injection=_python_context_injection(
-                argument.annotation, module, index, module_path
+            injection=(
+                "framework_injected" if argument.arg == selected
+                else "unresolved" if unresolved and identities[argument.arg] != "caller_supplied"
+                else "caller_supplied"
             ),
             json_type=(
                 None
@@ -2481,7 +2515,6 @@ def _parameter(
             ),
         )
 
-    arguments = node.args
     positional = [*arguments.posonlyargs, *arguments.args]
     defaults: list[ast.expr | None] = [None] * (
         len(positional) - len(arguments.defaults)
@@ -2500,7 +2533,7 @@ def _parameter(
     # ``*args`` and ``**kwargs`` are deliberately absent: they are not schema
     # properties, and a tool that has them takes arguments this reader cannot
     # enumerate rather than one parameter named ``kwargs``.
-    return tuple(parameters)
+    return tuple(parameters), unresolved
 
 
 def _python_annotation(node: ast.expr | None) -> str | None:
@@ -2760,11 +2793,7 @@ def _python_class_identity(
         if named is None:
             return "unresolved"
         full = f"{named}.{rest}" if rest else named
-        return (
-            "framework_injected"
-            if _is_python_context_symbol(full, attribute)
-            else "caller_supplied"
-        )
+        return _python_imported_class_identity(full, attribute)
     if not isinstance(node, ast.Name):
         return "unresolved"
     resolved = module.binding_of(node.id, scope)
@@ -2789,32 +2818,46 @@ def _python_class_identity(
             # carries server names, not annotation provenance. Until that
             # evidence is read, preserve the injection question (#541 review).
             return "unresolved"
-        return (
-            "framework_injected"
-            if _is_python_context_symbol(imported.module, imported.symbol)
-            else "caller_supplied"
-        )
+        return _python_imported_class_identity(imported.module, imported.symbol)
     if isinstance(binding, ast.ClassDef):
-        if id(binding) in seen:
+        if id(binding) in seen or len(seen) >= _MAX_ANNOTATION_DEPTH:
             # Mutual bases. ``class A(B)`` beside ``class B(A)`` raises at
             # import time, but it parses, and a reader that followed it would
             # not return — so the cycle is an answer this reader does not have.
             return "unresolved"
+        if binding.decorator_list or binding.keywords:
+            # A decorator or metaclass can replace the class or alter ancestry.
+            return "unresolved"
         seen = seen | {id(binding)}
         identity: ContextInjection = "caller_supplied"
         for base in binding.bases:
-            head = base.value if isinstance(base, ast.Subscript) else base
-            resolved_base = _python_class_identity(head, module, index, module_path, base, seen)
-            if resolved_base == "framework_injected":
-                return "framework_injected"
+            if isinstance(base, ast.Subscript):
+                return "unresolved"
+            resolved_base = _python_class_identity(base, module, index, module_path, base, seen)
             if resolved_base == "unresolved":
-                identity = "unresolved"
+                return "unresolved"
+            if resolved_base == "framework_injected":
+                identity = "framework_injected"
         return identity
     # A ``def``, a parameter, an assignment: whatever the name refers to, this
     # reader did not read a class definition for it.
     return "unresolved"
 
 
+def _python_imported_class_identity(module: str, symbol: str) -> ContextInjection:
+    if _is_python_context_symbol(module, symbol):
+        return "framework_injected"
+    # A named supported library class is evidence; an arbitrary external
+    # package path is not. It can re-export Context or one of its subclasses.
+    if (module, symbol) in {("pydantic", "BaseModel"), ("pydantic.main", "BaseModel")}:
+        return "caller_supplied"
+    if module in _PYTHON_TYPING_MODULES and symbol in (
+        _PYTHON_BUILTIN_TYPE_NAMES | set(PYTHON_ANNOTATION_JSON_TYPES) | {"Any", "LiteralString", "Never", "NoReturn"}
+    ):
+        return "caller_supplied"
+    return "unresolved"
+
+
 def _python_json_type(
     node: ast.expr | None,
     module: _PythonModule,
@@ -2889,32 +2932,72 @@ def _python_context_injection(
     module: _PythonModule,
     index: PythonServerIndex,
     module_path: str | None,
+    *,
+    scope: ast.AST | None = None,
+    depth: int = 0,
 ) -> ContextInjection:
-    """Whether the framework supplies this parameter instead of the caller."""
+    """A bounded common injection profile, before whole-signature selection.
+
+    Preserve generic arguments and Annotated metadata: the framework resolves
+    them too. A scalar/context class, nullable union, or a container of proven
+    caller types is readable here; generic Context membership differs between
+    SDK and standalone FastMCP and remains an explicit local limitation.
+    """
 
     if annotation is None:
         # The SDK identifies the context by resolving a *type hint*; a
         # parameter without one is never injected, whatever it is named.
         return "caller_supplied"
-    members = _python_annotation_members(annotation, module, annotation)
-    if members is None:
+    if depth > _MAX_ANNOTATION_DEPTH:
         return "unresolved"
-    identities = {
-        _python_class_identity(
-            member.value if isinstance(member, ast.Subscript) else member,
-            module,
-            index,
-            module_path,
-            annotation,
-            frozenset(),
+    scope = annotation if scope is None else scope
+
+    def resolve(child: ast.expr) -> ContextInjection:
+        return _python_context_injection(
+            child, module, index, module_path, scope=scope, depth=depth + 1
         )
-        for member in members
-    }
-    if "framework_injected" in identities:
-        return "framework_injected"
+
+    if isinstance(annotation, ast.Constant):
+        if annotation.value is None:
+            return "caller_supplied"
+        if isinstance(annotation.value, str):
+            parsed = _parse_python_expression(annotation.value)
+            return "unresolved" if parsed is None else resolve(parsed)
+        return "unresolved"
+    if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
+        identities = {resolve(annotation.left), resolve(annotation.right)}
+    elif isinstance(annotation, ast.Subscript):
+        symbol = _python_annotation_symbol(annotation.value, module, scope)
+        elements = list(annotation.slice.elts) if isinstance(annotation.slice, ast.Tuple) else [annotation.slice]
+        if symbol == "Annotated":
+            if len(elements) < 2 or not all(isinstance(item, ast.Constant) for item in elements[1:]):
+                return "unresolved"
+            return resolve(elements[0])
+        if symbol == "Literal":
+            return "caller_supplied" if all(isinstance(item, ast.Constant) for item in elements) else "unresolved"
+        if symbol in {"Union", "Optional"}:
+            if not elements or (symbol == "Optional" and len(elements) != 1):
+                return "unresolved"
+            identities = {resolve(item) for item in elements}
+        elif symbol in {"list", "List", "dict", "Dict", "tuple", "Tuple", "set", "Set", "frozenset", "FrozenSet"}:
+            # typing aliases enforce arity during get_type_hints; builtin
+            # GenericAlias does not. An invalid alias anywhere prevents the
+            # SDK from resolving even an otherwise proven Context parameter.
+            required_arity = {"List": 1, "Set": 1, "FrozenSet": 1, "Dict": 2}.get(symbol)
+            if required_arity is not None and len(elements) != required_arity:
+                return "unresolved"
+            identities = {
+                "caller_supplied" if isinstance(item, ast.Constant) and item.value is Ellipsis else resolve(item)
+                for item in elements
+            }
+            return "caller_supplied" if identities == {"caller_supplied"} else "unresolved"
+        else:
+            return "unresolved"
+    else:
+        return _python_class_identity(annotation, module, index, module_path, scope, frozenset())
     if "unresolved" in identities:
         return "unresolved"
-    return "caller_supplied"
+    return "framework_injected" if "framework_injected" in identities else "caller_supplied"
 
 
 class _PythonOffsets:

src/agents_shipgate/inputs/mcp_server_source.py

@@ -452,6 +452,10 @@ def _tool_from_site(
 #: tool says which question it could not answer (#539).
 SURFACE_GAP_UNRESOLVED_CONTEXT = "unresolved_context_identity"
 
+# The framework selects over the whole signature, including return/variadic
+# annotations that have no published parameter row.
+SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE = "unresolved_context_signature"
+
 #: A parameter carrying no annotation at all. The type it publishes would be
 #: the emitter's fallback rather than anything the source said.
 SURFACE_GAP_UNTYPED_PARAMETER = "untyped_parameter"
@@ -512,7 +516,9 @@ def _signature_gaps(site: RegistrationSite) -> list[str]:
 
     if site.parameters is None:
         return []
-    gaps: set[str] = set()
+    gaps: set[str] = (
+        {SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE} if site.context_injection_unresolved else set()
+    )
     for parameter in site.parameters:
         if parameter.injection == "framework_injected":
             # Not published, so nothing about its type is claimed either.

tests/mcp_idiom_corpus.py

@@ -1715,6 +1715,47 @@ def __init__(
     # matching the last token of the spelling answers wrongly in *both*
     # directions, so each shape below is paired with the one that looks
     # identical and must come back the other way.
+    "python_context_first_match": SourceCase(
+        "python_context_first_match", "python",
+        "from mcp.server.fastmcp import FastMCP, Context\n"
+        "mcp = FastMCP('fixture')\n"
+        "@mcp.tool()\n"
+        "def lookup(first: Context, *, second: Context) -> str:\n"
+        "    return 'fixture'\n",
+    ),
+    "python_context_unknown_return": SourceCase(
+        "python_context_unknown_return", "python",
+        "from mcp.server.fastmcp import FastMCP, Context\n"
+        "mcp = FastMCP('fixture')\n"
+        "@mcp.tool()\n"
+        "def lookup(ctx: Context) -> Missing:\n"
+        "    return 'fixture'\n",
+    ),
+    "python_context_invalid_typing_arity": SourceCase(
+        "python_context_invalid_typing_arity", "python",
+        "from mcp.server.fastmcp import FastMCP, Context\n"
+        "from typing import List\n"
+        "mcp = FastMCP('fixture')\n"
+        "@mcp.tool()\n"
+        "def lookup(ctx: Context, payload: 'List[str, int]') -> str:\n"
+        "    return 'fixture'\n",
+    ),
+    "python_context_generic_limit": SourceCase(
+        "python_context_generic_limit", "python",
+        "from mcp.server.fastmcp import FastMCP, Context\n"
+        "mcp = FastMCP('fixture')\n"
+        "@mcp.tool()\n"
+        "def lookup(holder: list[Context]) -> str:\n"
+        "    return 'fixture'\n",
+    ),
+    "python_context_unknown_variadic": SourceCase(
+        "python_context_unknown_variadic", "python",
+        "from mcp.server.fastmcp import FastMCP\n"
+        "mcp = FastMCP('fixture')\n"
+        "@mcp.tool()\n"
+        "def lookup(query: str, **options: Missing) -> str:\n"
+        "    return 'fixture'\n",
+    ),
     "python_application_model_named_context": SourceCase(
         "python_application_model_named_context",
         "python",