From 87692c2886670e5eedf534798d44f906ae355d3b Mon Sep 17 00:00:00 2001 From: Pengfei Hu Date: Tue, 8 Sep 2026 20:01:10 -0700 Subject: [PATCH 1/2] Bound FastMCP injection to whole-signature evidence --- docs/mcp-registration-idioms.md | 42 ++++- src/agents_shipgate/inputs/mcp_idioms.py | 157 ++++++++++++----- .../inputs/mcp_server_source.py | 8 +- tests/mcp_idiom_corpus.py | 32 ++++ tests/test_fastmcp_injection_contract.py | 122 ++++++++++++++ tests/test_mcp_idioms.py | 14 ++ tests/test_mcp_server_source.py | 47 ++++-- tests/test_zero_install_detector.py | 1 + tools/shipgate-detect.py | 159 +++++++++++++----- 9 files changed, 482 insertions(+), 100 deletions(-) create mode 100644 tests/test_fastmcp_injection_contract.py diff --git a/docs/mcp-registration-idioms.md b/docs/mcp-registration-idioms.md index b5df8a31..53f488d2 100644 --- a/docs/mcp-registration-idioms.md +++ b/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 diff --git a/src/agents_shipgate/inputs/mcp_idioms.py b/src/agents_shipgate/inputs/mcp_idioms.py index 0fd7676f..895b94dc 100644 --- a/src/agents_shipgate/inputs/mcp_idioms.py +++ b/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,7 +2466,34 @@ 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: @@ -2469,8 +2501,10 @@ def _parameter( 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,66 @@ 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 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"}: + 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: diff --git a/src/agents_shipgate/inputs/mcp_server_source.py b/src/agents_shipgate/inputs/mcp_server_source.py index 9e6f324a..f0b3c688 100644 --- a/src/agents_shipgate/inputs/mcp_server_source.py +++ b/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. diff --git a/tests/mcp_idiom_corpus.py b/tests/mcp_idiom_corpus.py index dd10faf9..1eec2cbe 100644 --- a/tests/mcp_idiom_corpus.py +++ b/tests/mcp_idiom_corpus.py @@ -1715,6 +1715,38 @@ 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_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", diff --git a/tests/test_fastmcp_injection_contract.py b/tests/test_fastmcp_injection_contract.py new file mode 100644 index 00000000..4b30c357 --- /dev/null +++ b/tests/test_fastmcp_injection_contract.py @@ -0,0 +1,122 @@ +"""Static injection claims stay within the pinned common framework profile. + +SDK probes below call only the installed framework's signature utility on +functions authored in this test. No scanned server is imported or executed. +""" + +import pytest + +from agents_shipgate.inputs.mcp_idioms import scan_source + + +def _parameters(signature, *, imports="", definitions="", family="mcp.server.fastmcp"): + source = ( + f"from {family} import FastMCP, Context\n" + "from typing import Optional, Union, Annotated\n" + f"{imports}\n{definitions}\n" + "server = FastMCP('fixture')\n" + "@server.tool()\n" + f"def lookup({signature}:\n return 'fixture'\n" + ) + result = scan_source(source, "python") + assert not result.anomalies + assert len(result.sites) == 1 and result.sites[0].name == "lookup" + return {parameter.name: parameter.injection for parameter in result.sites[0].parameters} + + +@pytest.mark.parametrize("family", ["mcp.server.fastmcp", "fastmcp"]) +@pytest.mark.parametrize("annotation", ["Context", "Context | None", "Optional[Context]", '"Context"', "Annotated[Context, 'description']"]) +def test_single_canonical_context_is_injected(family, annotation): + assert _parameters(f"query: str, ctx: {annotation}) -> str", family=family) == { + "query": "caller_supplied", "ctx": "framework_injected", + } + + +@pytest.mark.parametrize("family", ["mcp.server.fastmcp", "fastmcp"]) +def test_first_matching_context_is_the_only_excluded_parameter(family): + assert _parameters("first: Context, second: Context) -> str", family=family) == { + "first": "framework_injected", "second": "caller_supplied", + } + + +@pytest.mark.parametrize("annotation", [ + "list[Context]", "Context[object, None]", "list[list[Context]]", + "Optional[Context[object, None]]", +]) +@pytest.mark.parametrize("family", ["mcp.server.fastmcp", "fastmcp"]) +def test_generic_context_has_an_explicit_local_limit(annotation, family): + assert _parameters(f"ctx: {annotation}) -> str", family=family)["ctx"] == "unresolved" + + +@pytest.mark.parametrize("signature", [ + "ctx: Context, payload: Missing) -> str", + "ctx: Context) -> Missing", + "ctx: Context, *values: Missing) -> str", + "ctx: Context, **values: Missing) -> str", + "ctx: Context, payload: list[Missing]) -> str", + "ctx: Context, payload: Annotated[str, missing_metadata]) -> str", + "ctx: Context, payload: Annotated[str, Missing()]) -> str", + "*values: Context, ctx: Context) -> str", +]) +def test_unresolved_whole_signature_does_not_hide_context(signature): + assert _parameters(signature)["ctx"] == "unresolved" + + +@pytest.mark.parametrize("imports, annotation", [ + ("from acme.models import Context as ExternalContext", "ExternalContext"), + ("import acme.models", "acme.models.Context"), + ("from .models import Context as ExternalContext", "ExternalContext"), +]) +def test_external_class_path_does_not_establish_caller_ownership(imports, annotation): + assert _parameters(f"value: {annotation}) -> str", imports=imports)["value"] == "unresolved" + + +def test_application_model_and_ordinary_context_name_remain_caller_inputs(): + assert _parameters( + "context: str, payload: ApplicationContext) -> str", + imports="from pydantic import BaseModel", + definitions="class ApplicationContext(BaseModel):\n account_id: str", + ) == {"context": "caller_supplied", "payload": "caller_supplied"} + + +def test_proven_local_context_subclass_preserves_injection(): + assert _parameters( + "ctx: Reporting) -> str", definitions="class Reporting(Context):\n pass", + ) == {"ctx": "framework_injected"} + + +@pytest.mark.parametrize("definition", [ + "@decorate\nclass Reporting(Context):\n pass", + "class Reporting(Context, metaclass=Custom):\n pass", + "class Reporting(Context, Missing):\n pass", + "class Reporting(Missing, Context):\n pass", +]) +def test_unresolved_local_class_construction_cannot_hide_a_parameter(definition): + assert _parameters("ctx: Reporting) -> str", definitions=definition) == {"ctx": "unresolved"} + + +def test_installed_sdk_whole_signature_and_generic_semantics(): + sdk = pytest.importorskip("mcp.server.fastmcp.utilities.context_injection") + from mcp.server.fastmcp import Context + + def direct(ctx: Context) -> str: + return "" + + def two(first: Context, second: Context) -> str: + return "" + + def generic(holder: list[Context]) -> str: + return "" + + def parameterized(ctx: Context[object, None]) -> str: + return "" + + def unresolved(ctx: Context, payload: "MissingFixtureType") -> str: # noqa: F821 + return "" + + assert sdk.find_context_parameter(direct) == "ctx" + assert sdk.find_context_parameter(two) == "first" + assert sdk.find_context_parameter(generic) == "holder" + # The SDK Context is a Pydantic generic: parameterization materializes a class. + assert sdk.find_context_parameter(parameterized) == "ctx" + assert sdk.find_context_parameter(unresolved) is None diff --git a/tests/test_mcp_idioms.py b/tests/test_mcp_idioms.py index cf3fc475..3b71b282 100644 --- a/tests/test_mcp_idioms.py +++ b/tests/test_mcp_idioms.py @@ -378,6 +378,20 @@ def test_a_python_tool_with_no_parameters_is_not_one_without_a_signature(): assert site.returns is None +@pytest.mark.parametrize("case, remaining, limited", [ + ("python_context_first_match", ["second"], False), + ("python_context_unknown_return", ["ctx"], True), + ("python_context_generic_limit", ["holder"], True), + ("python_context_unknown_variadic", ["query"], True), +]) +def test_context_selection_and_limit_share_the_idiom_corpus(case, remaining, limited): + fixture = REGRESSIONS[case] + site = scan_source(fixture.text, fixture.language).sites[0] + assert site.name == "lookup" + assert [p.name for p in site.parameters if p.injection != "framework_injected"] == remaining + assert site.context_injection_unresolved is limited + + def test_a_python_variadic_is_not_a_signature_parameter(): """``*args`` and ``**kwargs`` name no property a caller can send. diff --git a/tests/test_mcp_server_source.py b/tests/test_mcp_server_source.py index 4e204530..010e0b25 100644 --- a/tests/test_mcp_server_source.py +++ b/tests/test_mcp_server_source.py @@ -1282,8 +1282,12 @@ def test_a_conventional_parameter_name_is_still_a_tool_input(tmp_path): "context", "runtime", ] - # Only the annotated context is injected. `list[Context]` is a list. - assert [p.name for p in by_name["search"].parameters] == ["query", "holder"] + # Generic membership is outside the shared profile. It prevents a whole- + # signature selection; no independently classified Context disappears. + assert [p.name for p in by_name["search"].parameters] == [ + "query", "reporter", "optional", "parameterised", "holder", + ] + assert mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE in by_name["search"].extraction["surface_gaps"] def _corpus_workspace(tmp_path, case_name: str, *, name: str) -> Path: @@ -1302,6 +1306,23 @@ def _corpus_workspace(tmp_path, case_name: str, *, name: str) -> Path: return workspace +@pytest.mark.parametrize("case, remaining, limited", [ + ("python_context_first_match", ["second"], False), + ("python_context_unknown_return", ["ctx"], True), + ("python_context_generic_limit", ["holder"], True), + ("python_context_unknown_variadic", ["query"], True), +]) +def test_whole_signature_limit_survives_the_production_loader(tmp_path, case, remaining, limited): + workspace = _corpus_workspace(tmp_path, case, name=case) + tool = load_mcp_server_source(_source("server.py"), workspace).tools[0] + assert tool.name == "lookup" + assert [p.name for p in tool.parameters] == remaining + assert tool.extraction_confidence == mcp_server_source.EXTRACTION_CONFIDENCE + assert (mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE in tool.extraction.get("surface_gaps", [])) is limited + if limited: + assert tool.extraction["surface"] == SURFACE_PARTIAL + + def test_an_application_model_named_context_stays_a_required_input(tmp_path): """The framework injects on the annotation's *binding*, not its spelling. @@ -1498,6 +1519,8 @@ def test_a_signature_publishes_the_type_the_annotation_denotes(tmp_path): assert tool.input_schema["properties"]["untyped"] == {} assert tool.extraction["surface_gaps"] == [ mcp_server_source.SURFACE_GAP_UNREPRESENTABLE_ANNOTATION, + mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT, + mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE, mcp_server_source.SURFACE_GAP_UNTYPED_PARAMETER, ] # Every published surface describes the same evidence: a parameter is in @@ -1544,6 +1567,7 @@ def test_a_container_names_its_kind_and_a_mapping_needs_string_keys(tmp_path): assert tool.extraction["surface_gaps"] == [ mcp_server_source.SURFACE_GAP_UNREPRESENTABLE_ANNOTATION, mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT, + mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE, ] @@ -1576,6 +1600,7 @@ def test_a_spelling_means_what_it_looks_like_only_while_nothing_rebinds_it( assert tool.extraction["surface_gaps"] == [ mcp_server_source.SURFACE_GAP_UNREPRESENTABLE_ANNOTATION, mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT, + mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE, ] @@ -1611,17 +1636,8 @@ def test_a_union_denotes_a_type_only_when_its_arms_agree(tmp_path): } -def test_a_class_imported_from_another_package_is_a_caller_input(tmp_path): - """An absolute import outside the framework's own packages settles it. - - The bound is written where it is taken: a class defined in another - distribution could subclass the framework's context and be injected too, - and reading that would mean reading that distribution. The parameter is - published either way — what would differ is one `required` flag, on a - shape none of the surveyed servers writes — while treating every imported - model as unresolved would put a question mark on the ordinary Pydantic - parameter every server has. - """ +def test_an_external_import_without_class_provenance_remains_unresolved(tmp_path): + """An external package can export Context or a subclass; its path is not proof.""" workspace = _corpus_workspace( tmp_path, "python_context_from_another_package", name="package" @@ -1632,7 +1648,7 @@ def test_a_class_imported_from_another_package_is_a_caller_input(tmp_path): assert tool.input_schema["required"] == ["context"] assert ( mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT - not in tool.extraction["surface_gaps"] + in tool.extraction["surface_gaps"] ) @@ -1704,7 +1720,8 @@ def test_an_unreadable_return_annotation_is_not_a_string_output_schema(tmp_path) assert by_name["modelled"].output_schema == {} assert by_name["modelled"].extraction["surface_gaps"] == [ - mcp_server_source.SURFACE_GAP_UNREPRESENTABLE_ANNOTATION + mcp_server_source.SURFACE_GAP_UNREPRESENTABLE_ANNOTATION, + mcp_server_source.SURFACE_GAP_UNRESOLVED_CONTEXT_SIGNATURE, ] assert by_name["counted"].output_schema == {"type": "number"} assert "surface_gaps" not in by_name["counted"].extraction diff --git a/tests/test_zero_install_detector.py b/tests/test_zero_install_detector.py index 9a05d335..26f0b701 100644 --- a/tests/test_zero_install_detector.py +++ b/tests/test_zero_install_detector.py @@ -1682,6 +1682,7 @@ def _site_fields(site: Any) -> dict[str, Any]: ), "returns": site.returns, "returns_json_type": site.returns_json_type, + "context_injection_unresolved": site.context_injection_unresolved, "proves_server": site.proves_server, } diff --git a/tools/shipgate-detect.py b/tools/shipgate-detect.py index 206fdfba..7d28bf1c 100644 --- a/tools/shipgate-detect.py +++ b/tools/shipgate-detect.py @@ -1046,6 +1046,7 @@ class RegistrationSite: #: there is no annotation and when there is one this reader cannot #: represent; ``returns`` tells the two apart. returns_json_type: str | None = None + 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. False for #: every lexical idiom: those match a spelling. The Python idiom follows @@ -2763,6 +2764,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, @@ -2771,10 +2775,11 @@ def _python_site( span=span, description=_python_tool_description(node, decorator), unresolved_reason=unresolved, - # Withheld with the name: the schema comes from the same object. - parameters=( - None if wrapped else _python_signature(node, module, index, module_path) - ), + # 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=parameters, + context_injection_unresolved=context_injection_unresolved, returns=None if wrapped else _python_annotation(node.returns), returns_json_type=( None @@ -2854,7 +2859,34 @@ 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: @@ -2862,8 +2894,10 @@ def _parameter( 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 @@ -2874,7 +2908,6 @@ def _parameter( ), ) - arguments = node.args positional = [*arguments.posonlyargs, *arguments.args] defaults: list[ast.expr | None] = [None] * ( len(positional) - len(arguments.defaults) @@ -2893,7 +2926,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: @@ -3153,11 +3186,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) @@ -3182,32 +3211,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, @@ -3282,32 +3325,66 @@ 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 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"}: + 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: From 9493d77545e8756e11daebd17d67d7dc67490967 Mon Sep 17 00:00:00 2001 From: Pengfei Hu Date: Tue, 8 Sep 2026 20:18:09 -0700 Subject: [PATCH 2/2] Keep invalid typing arity visible in Context signatures --- src/agents_shipgate/inputs/mcp_idioms.py | 8 +++++- tests/mcp_idiom_corpus.py | 9 ++++++ tests/test_fastmcp_injection_contract.py | 36 ++++++++++++++++++++++++ tests/test_mcp_idioms.py | 1 + tests/test_mcp_server_source.py | 1 + tools/shipgate-detect.py | 8 +++++- 6 files changed, 61 insertions(+), 2 deletions(-) diff --git a/src/agents_shipgate/inputs/mcp_idioms.py b/src/agents_shipgate/inputs/mcp_idioms.py index 895b94dc..bbcc8d31 100644 --- a/src/agents_shipgate/inputs/mcp_idioms.py +++ b/src/agents_shipgate/inputs/mcp_idioms.py @@ -2976,10 +2976,16 @@ def resolve(child: ast.expr) -> ContextInjection: if symbol == "Literal": return "caller_supplied" if all(isinstance(item, ast.Constant) for item in elements) else "unresolved" if symbol in {"Union", "Optional"}: - if symbol == "Optional" and len(elements) != 1: + 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 diff --git a/tests/mcp_idiom_corpus.py b/tests/mcp_idiom_corpus.py index 1eec2cbe..26723241 100644 --- a/tests/mcp_idiom_corpus.py +++ b/tests/mcp_idiom_corpus.py @@ -1731,6 +1731,15 @@ def __init__( "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" diff --git a/tests/test_fastmcp_injection_contract.py b/tests/test_fastmcp_injection_contract.py index 4b30c357..890370ca 100644 --- a/tests/test_fastmcp_injection_contract.py +++ b/tests/test_fastmcp_injection_contract.py @@ -4,6 +4,8 @@ functions authored in this test. No scanned server is imported or executed. """ +import typing + import pytest from agents_shipgate.inputs.mcp_idioms import scan_source @@ -62,6 +64,40 @@ def test_unresolved_whole_signature_does_not_hide_context(signature): assert _parameters(signature)["ctx"] == "unresolved" +@pytest.mark.parametrize("annotation", [ + "List[str, int]", "List[()]", "Set[str, int]", "FrozenSet[str, int]", + "Dict[str]", "Dict[str, str, int]", "Union[()]", +]) +def test_invalid_typing_arity_cannot_hide_context(annotation): + sdk = pytest.importorskip("mcp.server.fastmcp.utilities.context_injection") + from mcp.server.fastmcp import Context + + # Authored test annotations only: get_type_hints must reach the same + # semantic refusal that stops the real SDK's whole-signature resolver. + def specimen(ctx, payload): + return "" + + specimen.__annotations__ = {"ctx": Context, "payload": f"typing.{annotation}", "return": str} + with pytest.raises(TypeError): + typing.get_type_hints(specimen) + assert sdk.find_context_parameter(specimen) is None + assert _parameters( + f"ctx: Context, payload: {annotation}) -> str", + imports="from typing import List, Set, FrozenSet, Dict", + ) == {"ctx": "unresolved", "payload": "unresolved"} + + +@pytest.mark.parametrize("annotation", [ + "List[str]", "Set[str]", "FrozenSet[str]", "Dict[str, int]", + "Union[str, int]", "list[str, int]", +]) +def test_supported_container_arity_keeps_context_injection(annotation): + assert _parameters( + f"ctx: Context, payload: {annotation}) -> str", + imports="from typing import List, Set, FrozenSet, Dict", + ) == {"ctx": "framework_injected", "payload": "caller_supplied"} + + @pytest.mark.parametrize("imports, annotation", [ ("from acme.models import Context as ExternalContext", "ExternalContext"), ("import acme.models", "acme.models.Context"), diff --git a/tests/test_mcp_idioms.py b/tests/test_mcp_idioms.py index 3b71b282..6401f6f6 100644 --- a/tests/test_mcp_idioms.py +++ b/tests/test_mcp_idioms.py @@ -381,6 +381,7 @@ def test_a_python_tool_with_no_parameters_is_not_one_without_a_signature(): @pytest.mark.parametrize("case, remaining, limited", [ ("python_context_first_match", ["second"], False), ("python_context_unknown_return", ["ctx"], True), + ("python_context_invalid_typing_arity", ["ctx", "payload"], True), ("python_context_generic_limit", ["holder"], True), ("python_context_unknown_variadic", ["query"], True), ]) diff --git a/tests/test_mcp_server_source.py b/tests/test_mcp_server_source.py index 010e0b25..3191f53f 100644 --- a/tests/test_mcp_server_source.py +++ b/tests/test_mcp_server_source.py @@ -1309,6 +1309,7 @@ def _corpus_workspace(tmp_path, case_name: str, *, name: str) -> Path: @pytest.mark.parametrize("case, remaining, limited", [ ("python_context_first_match", ["second"], False), ("python_context_unknown_return", ["ctx"], True), + ("python_context_invalid_typing_arity", ["ctx", "payload"], True), ("python_context_generic_limit", ["holder"], True), ("python_context_unknown_variadic", ["query"], True), ]) diff --git a/tools/shipgate-detect.py b/tools/shipgate-detect.py index 7d28bf1c..51611038 100644 --- a/tools/shipgate-detect.py +++ b/tools/shipgate-detect.py @@ -3369,10 +3369,16 @@ def resolve(child: ast.expr) -> ContextInjection: if symbol == "Literal": return "caller_supplied" if all(isinstance(item, ast.Constant) for item in elements) else "unresolved" if symbol in {"Union", "Optional"}: - if symbol == "Optional" and len(elements) != 1: + 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