Ver Fonte

feat(mcp): domain layers can register inputSchema

Lukas Goldschmidt há 1 mês atrás
pai
commit
c0230cdad5
1 ficheiros alterados com 24 adições e 3 exclusões
  1. 24 3
      virtuoso_mcp.py

+ 24 - 3
virtuoso_mcp.py

@@ -91,6 +91,14 @@ def mcp_result(id_value: Any, result: Dict[str, Any]) -> Dict[str, Any]:
 
 
 def _mcp_tool_definition(name: str) -> Dict[str, Any]:
+    if name in TOOL_SCHEMAS:
+        schema = TOOL_SCHEMAS[name]
+        # schema is expected to be an MCP-compatible inputSchema object.
+        return {
+            "name": name,
+            "description": TOOL_DOCS.get(name, ""),
+            "inputSchema": schema,
+        }
     # Incremental compliance step: add explicit input schemas for the most-used tools.
     # We still keep `additionalProperties: True` so we don't break existing clients.
     base: Dict[str, Any] = {
@@ -839,8 +847,14 @@ TOOLS = {
     "load_examples": tool_load_examples,
 }
 
+# Tool input schemas registered by domain layers (e.g., garden_layer).
+TOOL_SCHEMAS: Dict[str, Any] = {}
+
 
-def load_domain_layers(tools: Dict[str, Callable[[Dict[str, Any]], Any]]) -> None:
+def load_domain_layers(
+    tools: Dict[str, Callable[[Dict[str, Any]], Any]],
+    tool_schemas: Dict[str, Any],
+) -> None:
     raw = os.getenv("DOMAIN_LAYERS", "garden_layer.plugin")
     modules = [item.strip() for item in raw.split(",") if item.strip()]
     if not modules:
@@ -871,12 +885,19 @@ def load_domain_layers(tools: Dict[str, Callable[[Dict[str, Any]], Any]]) -> Non
             logger.warning("Domain layer '%s' does not expose register_layer", module_name)
             continue
         try:
-            register(tools)
+            # Domain layer may optionally register input schemas.
+            try:
+                if register.__code__.co_argcount >= 2:
+                    register(tools, tool_schemas)
+                else:
+                    register(tools)
+            except Exception:
+                register(tools)
             logger.info("Loaded domain layer '%s'", module_name)
         except Exception as exc:
             logger.exception("Domain layer '%s' failed to register: %s", module_name, exc)
 
-load_domain_layers(TOOLS)
+load_domain_layers(TOOLS, TOOL_SCHEMAS)
 
 TOOL_DOCS = {
     "sparql_query": "Execute a bounded SELECT query and return the JSON result.",