[Deep Agent 13/15] Curated Utils-Server tools via the Jarvis Registry MCP gateway (#47)

agent/README.md

@@ -69,3 +69,23 @@ Every deep-agent run is **one Langfuse trace** built by `deep_agent.observabilit
 - run-level scores: `deep-agent.model_calls`, `.tool_calls`, `.sql_executions`, `.executed_successfully`, `.final_sql_present`, `.latency_s`
 
 In Langfuse, filter traces by **tag = `deep-agent`** (save it as a view). Disable tracing with `DEEP_AGENT_TRACING_ENABLED=false`.
+
+### Utils-Server tools (deep agent, optional)
+
+A curated subset of [Utils-Server](https://github.com/matzpen-agency/Utils-Server) tools is loaded from the Jarvis Registry
+MCP gateway at startup (`deep_agent/utils_mcp.py`). Set:
+
+```bash
+JARVIS_GATEWAY_URL=http://jarvis-mcp-gateway:8001   # unset = disabled
+JARVIS_MCP_NAMESPACE=global                         # gateway namespace exposing Utils-Server
+JARVIS_API_KEY=...                                  # sent as "Authorization: ApiKey <key>"
+```
+
+| Agent | Tools |
+|---|---|
+| main | `datetime_parse_relative_expression`, `text_normalize`, `trino_validate_spatial_expression` |
+| value-resolver | `datetime_parse_relative_expression`, `text_normalize`, `geo_create_buffer`, `geo_validate_geometry` |
+
+The registry service identity needs the `datetime:read`, `text:read`, `geo:read` and `trino:read` scopes, and each tool must be
+approved in the registry. SQL is never executed through Utils-Server. If the gateway is unreachable the agent starts without
+these tools and logs a warning.

agent/pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "deepagents>=0.6.12",
     "urllib3>=2.7.0",
     "pyyaml>=6.0",
+    "langchain-mcp-adapters>=0.2,<0.3",
 ]
 
 [project.scripts]

agent/src/agent/config.py

@@ -67,6 +67,11 @@ class AgentSettings(BaseSettings):
     DEEP_AGENT_PROMPT_CACHE_TTL_S: int = Field(default=60, ge=0)
     DEEP_AGENT_SUBAGENT_MODEL: str | None = None    # smaller/faster model for subagents (default: main model)
     DEEP_AGENT_SUBAGENT_MAX_TOOL_CALLS: int = Field(default=15, gt=0)
+    # Utils-Server tools via Jarvis Registry MCP gateway (see deep_agent/utils_mcp.py); unset URL disables
+    JARVIS_GATEWAY_URL: str | None = None           # e.g. http://jarvis-mcp-gateway:8001
+    JARVIS_MCP_NAMESPACE: str = "global"            # gateway namespace (provider slug) exposing Utils-Server
+    JARVIS_API_KEY: str = ""                        # sent as "Authorization: ApiKey <key>"
+    UTILS_MCP_TIMEOUT_S: float = Field(default=10.0, gt=0)
 
     # ── Trino connection info for explicit catalog/schema qualification ──
     TRINO_CATALOG: str = ""

agent/src/deep_agent/agent.py

@@ -16,6 +16,7 @@
 from deep_agent.resources import build_resources
 from deep_agent.subagents import build_subagents
 from deep_agent.tools import DEEP_AGENT_TOOLS
+from deep_agent.utils_mcp import load_utils_tools
 
 
 # Packaged system prompt (the Langfuse-managed version, when available, is loaded per agent build and refreshed per call).
@@ -27,7 +28,12 @@ def _default_llm() -> BaseChatModel:
     return get_llm().model_copy(update={"request_timeout": settings.DEEP_AGENT_LLM_TIMEOUT_S, "max_retries": 0})
 
 
-def create_deep_agent(model: BaseChatModel | None = None):
+async def acreate_deep_agent(model: BaseChatModel | None = None):
+    """create_deep_agent plus the curated Utils-Server tools from the Jarvis Registry gateway (when configured)."""
+    return create_deep_agent(model=model, extra_tools=await load_utils_tools())
+
+
+def create_deep_agent(model: BaseChatModel | None = None, extra_tools: dict[str, list] | None = None):
     """
     Creates and returns a compiled LangGraph deep agent equipped with all exploration,
     enrichment, temporal, geospatial, and semantic validation tools.
@@ -37,17 +43,19 @@ def create_deep_agent(model: BaseChatModel | None = None):
 
     Args:
         model: Chat model override (tests); defaults to the configured LLM.
+        extra_tools: Additional tools per agent ("main" or a subagent name), e.g. from load_utils_tools().
     """
+    extra_tools = extra_tools or {}
     llm = model or _default_llm()
     memory = MemorySaver()
     resources = build_resources()
     system_prompt = load_prompt("system").text
 
     agent = create_harness_agent(
         model=llm,
-        tools=DEEP_AGENT_TOOLS,
+        tools=[*DEEP_AGENT_TOOLS, *extra_tools.get("main", [])],
         system_prompt=system_prompt,
-        subagents=build_subagents(llm),
+        subagents=build_subagents(llm, extra_tools),
         middleware=[
             DeepAgentContextMiddleware(),
             PromptRefreshMiddleware("system", system_prompt),

agent/src/deep_agent/benchmark.py

@@ -137,13 +137,14 @@ def main() -> None:
     args = parser.parse_args()
 
     from agent.utils.jeen_metadata_client import close_mcp_sessions
-    from deep_agent.agent import create_deep_agent
+    from deep_agent.agent import acreate_deep_agent
 
     questions = json.loads(args.questions.read_text(encoding="utf-8"))
 
     async def _run():
         try:
-            return await run_benchmark(create_deep_agent(), questions, DeepAgentContext(connection_id=args.connection_id),
+            return await run_benchmark(await acreate_deep_agent(), questions,
+                                       DeepAgentContext(connection_id=args.connection_id),
                                        concurrency=args.concurrency)
         finally:
             await close_mcp_sessions()

agent/src/deep_agent/cli.py

@@ -19,7 +19,7 @@
 from langchain_core.messages import BaseMessage
 
 from agent.config import settings
-from deep_agent.agent import create_deep_agent
+from deep_agent.agent import acreate_deep_agent
 from agent.utils.jeen_metadata_client import close_mcp_sessions
 from deep_agent import observability
 from deep_agent.answer import best_effort_result, render_markdown
@@ -246,7 +246,7 @@ def next_thread_id() -> str:
     print(f"{MAGENTA}{BOLD}================================================================{RESET}\n")
 
     print(f"{MAGENTA}{BOLD}Initializing Deep Agent...{RESET}")
-    agent = create_deep_agent()
+    agent = await acreate_deep_agent()
 
     if args.interactive:
         print(f"{GREEN}{BOLD}Starting Interactive Mode. Type 'quit', 'exit', or 'q' to stop.{RESET}")

agent/src/deep_agent/envelope.py

@@ -16,6 +16,7 @@
     "INVALID_INPUT": "Fix the tool arguments as described in the message and call again.",
     "ENRICHMENT_ERROR": "Enrichment failed; the original SQL is unchanged — verify filter values with search_column_values instead.",
     "LOCATION_NOT_FOUND": "The location could not be geocoded. Try the Hebrew or English official name, or filter by a text column instead.",
+    "EXTERNAL_TOOL_ERROR": "The Utils-Server utility failed — continue without it (fall back to the skill's rules).",
 }
 
 

agent/src/deep_agent/skills/spatial-analyst/SKILL.md

@@ -107,6 +107,11 @@ WHERE ST_Distance(to_spherical_geography(ST_Point(e.lon, e.lat)),
 If the landmark is only known by name, `resolve_location_to_wkt` gives the surrounding area polygon; use its
 centroid `ST_Centroid(ST_GeometryFromText(...))` as the point, and say so in `assumptions`.
 
+**If the Utils-Server tools are available:** check a spatial expression with `trino_validate_spatial_expression`
+before putting it in a query — it reports whether the expression is valid and the functions exist on this Trino (with
+close-name suggestions for typos), and can evaluate it on one row of a table. `geo_create_buffer` turns a point +
+radius (true metres) into a polygon returned as GeoJSON; use it in SQL with `from_geojson_geometry('<geojson>')`.
+
 ---
 
 ## 4. Proximity & Directional Intent Patterns

agent/src/deep_agent/skills/temporal-analyst/SKILL.md

@@ -101,6 +101,11 @@ The Israeli week starts on **Sunday**, but Trino's `date_trunc('week', ...)` ret
 
 If a month is given without a year, assume the most recent past occurrence relative to today and state it in `assumptions`.
 
+**If the `datetime_parse_relative_expression` tool is available** (Utils-Server), use it to resolve point-in-time
+phrases deterministically — "לפני 3 ימים", "מחר", "יום ראשון הבא", "בעוד שבועיים" — with `timezone: "Asia/Jerusalem"`.
+It returns a single instant, not a period: "חודש שעבר" comes back as *one month ago*, not last month's bounds, and
+period phrases such as "השבוע" are rejected. Use its date as the anchor, then build the period with the rules above.
+
 ---
 
 ## Grouping by period (SELECT / GROUP BY — functions are fine here)

agent/src/deep_agent/subagents.py

@@ -51,16 +51,18 @@ def _subagent_model(main_model: BaseChatModel) -> BaseChatModel | None:
     return main_model.model_copy(update={"model_name": settings.DEEP_AGENT_SUBAGENT_MODEL})
 
 
-def build_subagents(main_model: BaseChatModel) -> list[dict]:
+def build_subagents(main_model: BaseChatModel, extra_tools: dict[str, list] | None = None) -> list[dict]:
+    """Subagent specs; `extra_tools` maps subagent name -> additional tools (e.g. Utils-Server)."""
     model = _subagent_model(main_model)
+    extra_tools = extra_tools or {}
     subagents = []
     for name, tools in SUBAGENT_TOOLS.items():
         prompt = load_prompt(name).text
         spec = {
             "name": name,
             "description": SUBAGENT_DESCRIPTIONS[name],
             "system_prompt": prompt,
-            "tools": tools,
+            "tools": [*tools, *extra_tools.get(name, [])],
             "skills": [SKILLS_ROUTE],
             "middleware": [
                 DeepAgentContextMiddleware(),

agent/src/deep_agent/utils_mcp.py

@@ -0,0 +1,138 @@
+"""
+Curated Utils-Server tools, loaded through the Jarvis Registry MCP gateway.
+
+Deterministic utilities beat prompt rules where they apply (Hebrew text normalisation,
+relative-date anchors, spatial-expression validation, buffers around points). They are
+optional: with JARVIS_GATEWAY_URL unset, or the gateway unreachable, the agent starts
+without them and logs why.
+
+- Endpoint: {JARVIS_GATEWAY_URL}/mcp/{JARVIS_MCP_NAMESPACE}/ (streamable HTTP),
+  auth `Authorization: ApiKey <JARVIS_API_KEY>` (the gateway's API-key scheme).
+- Only allow-listed tools are used, and each goes only to the agent that needs it
+  (UTILS_TOOL_ALLOCATION). SQL never runs through Utils-Server: all execution stays on the
+  guarded execute_trino_query path.
+- Every tool is wrapped in the Deep Agent envelope ({ok, data, error}); gateway errors
+  ({"error", "status_code", "response_text"}) become ok=false with EXTERNAL_TOOL_ERROR.
+"""
+
+import asyncio
+import json
+import logging
+import re
+import time
+from datetime import timedelta
+from typing import Any
+
+from langchain_core.tools import BaseTool, StructuredTool
+
+from agent.config import settings
+from deep_agent.envelope import fail, ok
+
+logger = logging.getLogger(__name__)
+
+# Which agent gets which Utils-Server tool ("main" = the top-level Deep Agent).
+UTILS_TOOL_ALLOCATION: dict[str, tuple[str, ...]] = {
+    "main": ("datetime_parse_relative_expression", "text_normalize", "trino_validate_spatial_expression"),
+    "value-resolver": ("datetime_parse_relative_expression", "text_normalize", "geo_create_buffer",
+                       "geo_validate_geometry"),
+    "schema-scout": (),
+    "result-verifier": (),
+}
+ALLOWED_UTILS_TOOLS = frozenset(name for names in UTILS_TOOL_ALLOCATION.values() for name in names)
+
+
+def _normalize(name: str) -> str:
+    return re.sub(r"[^a-z0-9_]", "_", name.lower())
+
+
+def _canonical(gateway_name: str) -> str | None:
+    """Map a gateway tool name (possibly namespaced, e.g. 'utils__text_normalize') to an allow-listed name."""
+    norm = _normalize(gateway_name)
+    for name in ALLOWED_UTILS_TOOLS:
+        if norm == name or norm.endswith(f"_{name}"):
+            return name
+    return None
+
+
+def _parse(raw: Any) -> Any:
+    if isinstance(raw, tuple):  # (content, artifact)
+        raw = raw[0]
+    if isinstance(raw, list):  # content blocks
+        raw = "".join(b.get("text", "") if isinstance(b, dict) else str(b) for b in raw)
+    if isinstance(raw, str):
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError:
+            return raw
+    return raw
+
+
+def wrap_utils_tool(tool: BaseTool, name: str) -> BaseTool:
+    """Expose an MCP tool under its canonical name, returning the Deep Agent envelope."""
+
+    async def _call(**kwargs: Any) -> str:
+        start = time.perf_counter()
+        try:
+            payload = _parse(await tool.ainvoke(kwargs))
+        except Exception as exc:
+            logger.warning("Utils tool %s failed: %s", name, exc)
+            return fail("EXTERNAL_TOOL_ERROR", f"{name} failed: {exc}")
+        elapsed_ms = int((time.perf_counter() - start) * 1000)
+        if isinstance(payload, dict) and "error" in payload and ("status_code" in payload or len(payload) == 1):
+            return fail("EXTERNAL_TOOL_ERROR", str(payload.get("response_text") or payload["error"]),
+                        status_code=payload.get("status_code"), latency_ms=elapsed_ms)
+        logger.debug("Utils tool %s answered in %d ms", name, elapsed_ms)
+        return ok(payload)
+
+    return StructuredTool.from_function(
+        coroutine=_call,
+        name=name,
+        description=f"[Utils-Server] {tool.description}",
+        args_schema=tool.args_schema,
+    )
+
+
+def allocate(tools: list[BaseTool]) -> dict[str, list[BaseTool]]:
+    """Pick the allow-listed gateway tools and hand each agent its share."""
+    by_name: dict[str, BaseTool] = {}
+    for tool in tools:
+        name = _canonical(tool.name)
+        if name and name not in by_name:
+            by_name[name] = wrap_utils_tool(tool, name)
+    missing = ALLOWED_UTILS_TOOLS - set(by_name)
+    if missing:
+        logger.warning("Utils-Server tools not available on the gateway (pending approval?): %s", sorted(missing))
+    return {agent: [by_name[n] for n in names if n in by_name] for agent, names in UTILS_TOOL_ALLOCATION.items()}
+
+
+def gateway_connection() -> dict[str, Any] | None:
+    if not settings.JARVIS_GATEWAY_URL:
+        return None
+    connection: dict[str, Any] = {
+        "transport": "streamable_http",
+        "url": f"{settings.JARVIS_GATEWAY_URL.rstrip('/')}/mcp/{settings.JARVIS_MCP_NAMESPACE}/",
+        "timeout": timedelta(seconds=settings.UTILS_MCP_TIMEOUT_S),
+    }
+    if settings.JARVIS_API_KEY:
+        connection["headers"] = {"Authorization": f"ApiKey {settings.JARVIS_API_KEY}"}
+    return connection
+
+
+async def load_utils_tools() -> dict[str, list[BaseTool]]:
+    """Load and allocate the curated Utils-Server tools; empty allocation when unavailable."""
+    empty: dict[str, list[BaseTool]] = {agent: [] for agent in UTILS_TOOL_ALLOCATION}
+    connection = gateway_connection()
+    if connection is None:
+        logger.info("JARVIS_GATEWAY_URL not set — Utils-Server tools disabled")
+        return empty
+    try:
+        from langchain_mcp_adapters.client import MultiServerMCPClient
+
+        client = MultiServerMCPClient({"jarvis": connection})
+        tools = await asyncio.wait_for(client.get_tools(), timeout=settings.UTILS_MCP_TIMEOUT_S)
+    except Exception as exc:
+        logger.warning("Utils-Server tools unavailable (%s: %s) — continuing without them", type(exc).__name__, exc)
+        return empty
+    allocation = allocate(tools)
+    logger.info("Loaded Utils-Server tools: %s", {k: [t.name for t in v] for k, v in allocation.items() if v})
+    return allocation