fix(mcp): pass params_schema to MCP tools via Python signatures

Fixes #64025

The hermes-tools MCP server was fetching each tool's JSON schema
into params_schema but never passing it to FastMCP's add_tool(),
so all published tools had an empty **kwargs signature. MCP clients
couldn't see parameters and arguments were dropped at dispatch.

This fix:
- Adds _signature_from_schema() to convert JSON schemas to Python
  function signatures with type annotations
- Attaches the generated signature/annotations to each handler closure
  so FastMCP introspects the real parameter structure
- Filters out None values before dispatch to avoid forwarding unset
  optional parameters

Impact: web_search, browser automation, vision, and other Hermes tools
are now properly callable from the codex_app_server runtime.
This commit is contained in:
liuhao1024 2026-07-14 07:42:46 +08:00 committed by Teknium
parent e357b69a61
commit 3ad5876feb
2 changed files with 176 additions and 4 deletions

View file

@ -44,6 +44,7 @@ Spawned by: CodexAppServerSession.ensure_started() when the runtime is
from __future__ import annotations
import inspect
import json
import logging
import os
@ -52,6 +53,49 @@ from typing import Any, Optional
logger = logging.getLogger(__name__)
# JSON Schema type -> Python type mapping for signature generation
_JSON_TO_PY = {
"string": str,
"integer": int,
"number": float,
"boolean": bool,
"array": list,
"object": dict,
}
def _signature_from_schema(schema: dict | None) -> tuple[inspect.Signature, dict[str, type]]:
"""Build a Python function signature and annotations from a JSON schema.
Args:
schema: JSON Schema dict with "properties" and "required" keys.
Returns:
(signature, annotations_dict) where signature has KEYWORD_ONLY params
and annotations maps param names to Python types.
"""
props = (schema or {}).get("properties") or {}
required = set((schema or {}).get("required") or [])
params, annots = [], {}
for pname, pspec in props.items():
if pname.startswith("_"):
continue
py = _JSON_TO_PY.get((pspec or {}).get("type"), Any)
ann, default = (
(py, inspect.Parameter.empty)
if pname in required
else (Optional[py], None)
)
annots[pname] = ann
params.append(
inspect.Parameter(
pname, inspect.Parameter.KEYWORD_ONLY, annotation=ann, default=default
)
)
return inspect.Signature(params, return_annotation=str), annots
# Tools we expose. Each name MUST match a registered Hermes tool that
# `model_tools.handle_function_call()` can dispatch.
@ -159,20 +203,28 @@ def _build_server() -> Any:
# the result string. We use add_tool() for full control over the
# input schema (FastMCP's @tool() decorator inspects type hints,
# which we can't get from a JSON schema at runtime).
def _make_handler(tool_name: str):
def _make_handler(tool_name: str, schema: dict | None):
sig, annots = _signature_from_schema(schema)
def _dispatch(**kwargs: Any) -> str:
try:
return handle_function_call(tool_name, kwargs or {})
# Filter out None values before dispatch so unset optionals
# aren't forwarded to the handler.
args = {k: v for k, v in kwargs.items() if v is not None}
return handle_function_call(tool_name, args or {})
except Exception as exc:
logger.exception("tool %s raised", tool_name)
return json.dumps({"error": str(exc), "tool": tool_name})
_dispatch.__name__ = tool_name
_dispatch.__doc__ = description
_dispatch.__signature__ = sig
_dispatch.__annotations__ = {**annots, "return": str}
return _dispatch
try:
mcp.add_tool(
_make_handler(name),
_make_handler(name, params_schema),
name=name,
description=description,
# FastMCP accepts JSON schema directly via the
@ -181,7 +233,7 @@ def _build_server() -> Any:
)
except TypeError:
# Older mcp SDK signature — fall back to decorator-style.
handler = _make_handler(name)
handler = _make_handler(name, params_schema)
handler = mcp.tool(name=name, description=description)(handler)
exposed_count += 1

View file

@ -8,6 +8,126 @@ build helper assembles a server when the SDK is present.
from __future__ import annotations
import inspect
from typing import get_args
from agent.transports.hermes_tools_mcp_server import (
_signature_from_schema,
)
class TestSignatureFromSchema:
"""Test the JSON Schema -> Python signature conversion."""
def test_simple_required_string_param(self):
"""A required string param becomes str with no default."""
schema = {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
}
sig, annots = _signature_from_schema(schema)
assert len(sig.parameters) == 1
param = sig.parameters["query"]
assert param.name == "query"
assert param.kind == inspect.Parameter.KEYWORD_ONLY
assert annots["query"] == str
assert param.default is inspect.Parameter.empty
def test_optional_integer_param(self):
"""An optional param gets Optional[type] with default=None."""
schema = {
"type": "object",
"properties": {"limit": {"type": "integer"}},
}
sig, annots = _signature_from_schema(schema)
param = sig.parameters["limit"]
# Optional[type] is type | None in Python 3.10+
assert param.default is None
def test_multiple_params_mixed_required_optional(self):
"""Mixed required and optional params are handled correctly."""
schema = {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer"},
"offset": {"type": "integer"},
},
"required": ["query"],
}
sig, annots = _signature_from_schema(schema)
assert len(sig.parameters) == 3
# query: required str
assert annots["query"] == str
assert sig.parameters["query"].default is inspect.Parameter.empty
# limit: optional int
assert sig.parameters["limit"].default is None
# offset: optional int
assert sig.parameters["offset"].default is None
def test_skip_private_params(self):
"""Params starting with '_' are excluded from the signature."""
schema = {
"type": "object",
"properties": {
"query": {"type": "string"},
"_internal": {"type": "string"},
},
"required": ["query", "_internal"],
}
sig, annots = _signature_from_schema(schema)
assert "_internal" not in sig.parameters
assert "_internal" not in annots
assert "query" in sig.parameters
def test_all_json_types(self):
"""All JSON schema types map to correct Python types."""
schema = {
"type": "object",
"properties": {
"s": {"type": "string"},
"i": {"type": "integer"},
"n": {"type": "number"},
"b": {"type": "boolean"},
"a": {"type": "array"},
"o": {"type": "object"},
},
"required": ["s", "i", "n", "b", "a", "o"],
}
sig, annots = _signature_from_schema(schema)
assert annots["s"] == str
assert annots["i"] == int
assert annots["n"] == float
assert annots["b"] == bool
assert annots["a"] == list
assert annots["o"] == dict
def test_empty_schema(self):
"""Empty schema returns empty signature."""
sig, annots = _signature_from_schema(None)
assert len(sig.parameters) == 0
assert len(annots) == 0
def test_return_annotation_is_str(self):
"""All generated signatures have str as return type."""
schema = {
"type": "object",
"properties": {"query": {"type": "string"}},
}
sig, annots = _signature_from_schema(schema)
assert sig.return_annotation == str