feat(tools): research action on net/fetch — cited answers over /v1/ask

Search finds pages, web_read reads one, research is the whole loop behind a
single door: the live cloud answer engine (api.hanzo.ai /v1/ask, mode=research)
plans queries, searches, reads the best pages and writes an answer that cites
them. It is an action on the same fetch tool, not a new package — search, read
and research are one concern (the web) and share one client and one auth.

The engine replies as server-sent events, which is the cloud's other reply
shape, so HanzoCloud gains stream() beside get/post: the base URL, the auth
header and the error mapping stay in the one place that already owns them.
Each `data:` frame is yielded as a typed event (status | sources | text |
follow_ups | done | error); the terminal [DONE] sentinel ends the stream and is
never an event. Frames accumulate into {answer, sources, follow_ups}; `deep`
normalizes to `research` (one pass, two names) and any other mode is refused
rather than silently downgraded.

hanzo-tools 0.3.4->0.3.5, -net 0.1.3->0.1.4, and -net's floor moves to
hanzo-tools>=0.3.5 so the action cannot install against a client with no
stream(). Tests replay a canned SSE stream through the real parser.

Co-authored-by: Hanzo Dev <[email protected]>

pkg/hanzo-tools-net/README.md

@@ -4,7 +4,12 @@ Network tools for Hanzo AI MCP (HIP-0300).
 
 ## Tools
 
-- `fetch` - HTTP fetch with content extraction (HTML/JSON/text)
+- `fetch` — the web, one door per verb:
+  - `search` — query → [url, title, snippet]
+  - `web_read` — url → clean markdown
+  - `research` — question → cited answer + sources (search + read + synthesize)
+  - `fetch` / `download` / `crawl` / `head` / `request` / `open`
+- `vision` — image (+ prompt) → text answer
 
 ## Installation
 

pkg/hanzo-tools-net/hanzo_tools/net/__init__.py

@@ -3,6 +3,7 @@
 Tools:
 - net: Unified network tool (HIP-0300)
   - search: Web search (Query → [URL, title, snippet])
+  - research: Cited answer over the live web (Query → {answer, sources})
   - fetch: Retrieve URL content (URL → {text, mime})
   - download: Save URL with assets (URL → Path)
   - crawl: Recursive site mirror (URL, depth → [Path])

pkg/hanzo-tools-net/hanzo_tools/net/fetch_tool.py

@@ -2,6 +2,7 @@
 
 This module provides the 'fetch' tool for network operations:
 - search: Query → [URL, title, snippet]
+- research: Query → {answer with citations, sources, follow_ups}
 - fetch: URL → {text, mime, status}
 - download: URL → Path (with assets)
 - crawl: URL + depth → [Path] (recursive mirror)
@@ -23,10 +24,16 @@ from hanzo_tools.core import (
     BaseTool,
     CloudError,
     HanzoCloud,
+    InvalidParamsError,
     ToolError,
     content_hash,
 )
 
+# A research pass legitimately runs for minutes — it plans, searches, reads a few
+# dozen pages and writes — so its stream gets its own budget instead of the
+# client's request-shaped default.
+RESEARCH_TIMEOUT = 300.0
+
 
 def _extract_markdown(resp: Any) -> str:
     """Best-effort markdown out of a Hanzo Crawl / Firecrawl response.
@@ -62,6 +69,32 @@ def _extract_markdown(resp: Any) -> str:
     return from_obj(payload)
 
 
+def _research_mode(mode: str) -> str:
+    """The one pass this action opens.
+
+    ``deep`` is the older name for the same research pass, so it normalizes.
+    Anything else is a different door — plain web search is its own action — and
+    is refused rather than silently downgraded.
+    """
+    m = (mode or "research").strip().lower()
+    if m in ("research", "deep"):
+        return "research"
+    raise InvalidParamsError(
+        f"unknown research mode: {mode}", param="mode", expected="research"
+    )
+
+
+def _source(s: Any) -> dict:
+    """A citation's card: where it is, what it is, and its mark."""
+    if not isinstance(s, dict):
+        return {"url": str(s), "title": "", "favicon": ""}
+    return {
+        "url": s.get("url", ""),
+        "title": s.get("title", ""),
+        "favicon": s.get("favicon", ""),
+    }
+
+
 def _crawl_metadata(resp: Any) -> dict:
     """Pull a metadata dict out of a crawl response, if present."""
     if isinstance(resp, dict):
@@ -107,6 +140,8 @@ Actions:
 - search: Web search — cloud Bing meta-search (api.hanzo.ai) with a local
   DuckDuckGo fallback (Query → [URL, title, snippet])
 - web_read: Fetch a URL → clean markdown (cloud Crawl4AI, local fallback)
+- research: Search + read + synthesize behind one door — the cloud answer
+  engine (Query → {answer with citations, sources, follow_ups})
 - fetch: Retrieve URL content (URL → {text, mime, status})
 - download: Save page with assets (URL → Path)
 - crawl: Recursive site mirror (URL, depth → [Path])
@@ -308,6 +343,77 @@ Effect: NONDETERMINISTIC_EFFECT (network I/O)
                 "status": response.status_code,
             }
 
+        @self.action("research", "Research a question → cited answer + sources")
+        async def research(
+            ctx: MCPContext,
+            query: str,
+            mode: str = "research",
+        ) -> dict:
+            """Research a question end to end and return a cited answer.
+
+            Search finds pages and web_read reads one; research is the whole
+            loop behind a single door — the cloud answer engine
+            (api.hanzo.ai /v1/ask) plans queries, searches, reads the best
+            pages and writes an answer that cites them. Its progress frames are
+            consumed here, so the caller gets the finished result without
+            orchestrating search + web_read itself.
+
+            Args:
+                query: The question to research
+                mode: research (default); "deep" names the same pass
+
+            Returns:
+                {answer, sources: [{url, title, favicon}], follow_ups, query, mode}
+
+            Effect: NONDETERMINISTIC_EFFECT
+            """
+            if not query or not query.strip():
+                raise InvalidParamsError("query is required", param="query")
+            mode = _research_mode(mode)
+            cloud = self._get_cloud()
+            if cloud is None:
+                raise ToolError(
+                    code="INVALID_PARAMS",
+                    message="Hanzo Cloud not configured: set HANZO_API_KEY "
+                    "or add .apiKey to ~/.hanzo/config.json",
+                )
+
+            answer = ""
+            deltas: list[str] = []
+            sources: list = []
+            follow_ups: list[str] = []
+            async for event in cloud.stream(
+                "/v1/ask", {"q": query, "mode": mode}, timeout=RESEARCH_TIMEOUT
+            ):
+                kind = event.get("type")
+                if kind == "text":
+                    deltas.append(event.get("delta") or "")
+                elif kind == "sources":
+                    sources = event.get("sources") or sources
+                elif kind == "follow_ups":
+                    follow_ups = event.get("questions") or []
+                elif kind == "done":
+                    answer = event.get("answer") or ""
+                    sources = event.get("sources") or sources
+                    break
+                elif kind == "error":
+                    raise ToolError(
+                        code="INTERNAL_ERROR",
+                        message=event.get("message") or "research stream failed",
+                        details={"query": query},
+                    )
+                # status frames are progress, not result
+
+            return {
+                # The done frame carries the finished text; the accumulated
+                # deltas are the same answer, and stand in if it arrives empty.
+                "answer": answer or "".join(deltas),
+                "sources": [_source(s) for s in sources],
+                "follow_ups": follow_ups,
+                "query": query,
+                "mode": mode,
+            }
+
         @self.action("fetch", "Retrieve URL content")
         async def fetch(
             ctx: MCPContext,

pkg/hanzo-tools-net/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "hanzo-tools-net"
-version = "0.1.3"
+version = "0.1.4"
 description = "Network tools for Hanzo AI (HIP-0300)"
 readme = "README.md"
 license = "MIT"
@@ -31,7 +31,7 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
-    "hanzo-tools>=0.3.0",
+    "hanzo-tools>=0.3.5",
     "mcp>=1.0.0",
     "httpx>=0.25.0",
 ]

pkg/hanzo-tools-net/tests/test_net_cloud.py

@@ -6,8 +6,9 @@ routing logic (cloud-first, local-fallback) is exercised deterministically.
 
 import asyncio
 import base64
+import json
 
-from hanzo_tools.core import CloudError
+from hanzo_tools.core import CloudError, HanzoCloud
 
 from hanzo_tools.net import FetchTool, VisionTool
 from hanzo_tools.net.fetch_tool import _crawl_metadata, _extract_markdown
@@ -154,6 +155,195 @@ def test_extract_markdown_shapes():
     assert _crawl_metadata({"data": {"metadata": {"title": "T"}}}) == {"title": "T"}
 
 
+# ── research (SSE over /v1/ask) ──────────────────────────────────────────
+#
+# These fake the wire, not the client: a real HanzoCloud runs over a canned
+# transport, so the SSE parser itself is under test — including the terminal
+# [DONE] sentinel, which is the end of the stream and never an event.
+
+SOURCE = {
+    "url": "https://clojure.org",
+    "title": "Clojure",
+    "snippet": "a dynamic Lisp",
+    "engine": "bing",
+    "favicon": "https://clojure.org/favicon.ico",
+}
+CARD = {
+    "url": "https://clojure.org",
+    "title": "Clojure",
+    "favicon": "https://clojure.org/favicon.ico",
+}
+
+
+def sse(*events):
+    """Canned SSE lines: one frame per event, then the [DONE] sentinel."""
+    lines = []
+    for e in events:
+        lines += [f"data: {json.dumps(e)}", ""]
+    return lines + ["data: [DONE]", ""]
+
+
+ANSWER_STREAM = sse(
+    {"type": "status", "stage": "searching", "detail": "who created clojure"},
+    {"type": "sources", "sources": [SOURCE]},
+    {"type": "text", "delta": "Rich Hickey "},
+    {"type": "text", "delta": "created [Clojure](https://clojure.org)."},
+    {"type": "follow_ups", "questions": ["When was Clojure released?"]},
+    {
+        "type": "done",
+        "answer": "Rich Hickey created [Clojure](https://clojure.org).",
+        "sources": [SOURCE],
+    },
+)
+
+
+class FakeSSEResponse:
+    def __init__(self, lines, status=200, body=""):
+        self.lines = lines
+        self.status_code = status
+        self.text = body
+
+    async def aiter_lines(self):
+        for line in self.lines:
+            yield line
+
+    async def aread(self):
+        return self.text.encode()
+
+
+class FakeStreamClient:
+    """An httpx-shaped client whose stream() replays canned SSE lines."""
+
+    def __init__(self, response):
+        self.response = response
+        self.calls = []
+
+    def stream(self, method, path, **kw):
+        self.calls.append((method, path, kw))
+        response = self.response
+
+        class Open:
+            async def __aenter__(self):
+                return response
+
+            async def __aexit__(self, *exc):
+                return False
+
+        return Open()
+
+
+def cloud_over(lines, status=200, body=""):
+    """A real HanzoCloud whose transport replays `lines`."""
+    cloud = HanzoCloud(key="hk-test")
+    client = FakeStreamClient(FakeSSEResponse(lines, status=status, body=body))
+
+    async def fake_get_client():
+        return client
+
+    cloud._get_client = fake_get_client
+    return cloud, client
+
+
+def researcher(lines, **kw):
+    cloud, client = cloud_over(lines, **kw)
+    t = FetchTool()
+    t._cloud = cloud
+    return t, client
+
+
+def test_stream_yields_events_and_skips_done_sentinel():
+    cloud, _ = cloud_over(ANSWER_STREAM)
+
+    async def collect():
+        return [e async for e in cloud.stream("/v1/ask", {"q": "x"})]
+
+    events = _run(collect())
+    assert [e["type"] for e in events] == [
+        "status",
+        "sources",
+        "text",
+        "text",
+        "follow_ups",
+        "done",
+    ]
+    assert all(isinstance(e, dict) for e in events)  # [DONE] never surfaced
+
+
+def test_research_parses_frames_into_answer_and_sources():
+    t, client = researcher(ANSWER_STREAM)
+    env = _run(t.call(None, action="research", query="who created clojure"))
+    assert env["ok"] is True
+    data = env["data"]
+    assert data["answer"] == "Rich Hickey created [Clojure](https://clojure.org)."
+    assert data["sources"] == [CARD]
+    assert data["follow_ups"] == ["When was Clojure released?"]
+    assert data["mode"] == "research"
+
+    method, path, kw = client.calls[0]
+    assert (method, path) == ("POST", "/v1/ask")
+    assert kw["json"] == {"q": "who created clojure", "mode": "research"}
+    assert kw["headers"]["Accept"] == "text/event-stream"
+
+
+def test_research_normalizes_deep_to_research():
+    t, client = researcher(ANSWER_STREAM)
+    env = _run(t.call(None, action="research", query="q", mode="deep"))
+    assert env["ok"] is True and env["data"]["mode"] == "research"
+    assert client.calls[0][2]["json"]["mode"] == "research"
+
+
+def test_research_refuses_another_mode():
+    t, client = researcher(ANSWER_STREAM)
+    env = _run(t.call(None, action="research", query="q", mode="news"))
+    assert env["ok"] is False and env["error"]["code"] == "INVALID_PARAMS"
+    assert client.calls == []
+
+
+def test_research_answer_falls_back_to_text_deltas():
+    # A done frame with no answer (a degraded synthesis) still returns the text
+    # the stream already delivered.
+    t, _ = researcher(
+        sse(
+            {"type": "sources", "sources": [SOURCE]},
+            {"type": "text", "delta": "Rich Hickey "},
+            {"type": "text", "delta": "created Clojure."},
+            {"type": "done", "answer": "", "sources": []},
+        )
+    )
+    env = _run(t.call(None, action="research", query="q"))
+    assert env["ok"] is True
+    assert env["data"]["answer"] == "Rich Hickey created Clojure."
+    assert env["data"]["sources"] == [CARD]  # last sources frame is kept
+
+
+def test_research_error_frame_surfaces():
+    t, _ = researcher(sse({"type": "error", "message": "upstream down"}))
+    env = _run(t.call(None, action="research", query="q"))
+    assert env["ok"] is False
+    assert env["error"]["code"] == "INTERNAL_ERROR"
+    assert "upstream down" in env["error"]["message"]
+
+
+def test_research_http_error_surfaces():
+    t, _ = researcher([], status=503, body="upstream unavailable")
+    env = _run(t.call(None, action="research", query="q"))
+    assert env["ok"] is False and "503" in env["error"]["message"]
+
+
+def test_research_not_configured_fails_closed():
+    t = FetchTool()
+    t._cloud = FakeCloud(configured=False)
+    env = _run(t.call(None, action="research", query="q"))
+    assert env["ok"] is False and env["error"]["code"] == "INVALID_PARAMS"
+
+
+def test_research_requires_a_query():
+    t, client = researcher(ANSWER_STREAM)
+    env = _run(t.call(None, action="research", query="  "))
+    assert env["ok"] is False and env["error"]["code"] == "INVALID_PARAMS"
+    assert client.calls == []
+
+
 # ── vision ───────────────────────────────────────────────────────────────
 
 

pkg/hanzo-tools/hanzo_tools/core/cloud.py

@@ -1,9 +1,9 @@
 """One client for the live Hanzo Cloud backend (api.hanzo.ai).
 
 Every hanzo-tools package that reaches the cloud — code index/search, vector
-search, web search, vision — talks to it through this single seam. There is
-exactly one place that knows the base URL, the auth header, and how to turn a
-non-2xx into a typed error; tools compose it, never re-implement it.
+search, web search, research, vision — talks to it through this one client.
+There is exactly one place that knows the base URL, the auth header, and how to
+turn a non-2xx into a typed error; tools compose it, never re-implement it.
 
 Auth resolves in order: ``HANZO_API_KEY`` env, the ``apiKey`` (hk- key) in
 ``~/.hanzo/config.json``, then the ``hanzo`` CLI's live IAM session. Base URL is
@@ -17,6 +17,7 @@ import json
 import subprocess
 from typing import Any, ClassVar
 from pathlib import Path
+from collections.abc import AsyncIterator
 
 DEFAULT_BASE = "https://api.hanzo.ai"
 
@@ -175,6 +176,51 @@ class HanzoCloud:
         """POST json_body to path, returning parsed JSON. Raises on failure."""
         return await self._request("POST", path, json=json_body or {})
 
+    async def stream(
+        self,
+        path: str,
+        json_body: dict[str, Any] | None = None,
+        timeout: float | None = None,
+    ) -> AsyncIterator[dict[str, Any]]:
+        """POST json_body to path and yield the events the server streams back.
+
+        Server-sent events are the cloud's other reply shape. The answer engine
+        (``/v1/ask``) writes ``data: {…}`` frames as it plans, searches, reads
+        and writes, so a caller follows the work instead of waiting minutes in
+        silence. It belongs beside get/post: the base URL, the auth header and
+        the error mapping stay in this one place.
+
+        Each frame's JSON object is yielded as a dict. The terminal ``[DONE]``
+        sentinel (the OpenAI convention the server mirrors) marks the end of the
+        stream, not an event, so it is skipped rather than handed to callers.
+        ``timeout`` overrides the client default for one call — a research pass
+        legitimately outlives a request-shaped budget.
+        """
+        client = await self._get_client()
+        headers = {**self._headers(), "Accept": "text/event-stream"}
+        try:
+            async with client.stream(
+                "POST",
+                path,
+                headers=headers,
+                json=json_body or {},
+                timeout=timeout if timeout is not None else self.timeout,
+            ) as resp:
+                if resp.status_code >= 400:
+                    await resp.aread()
+                    raise CloudError(
+                        f"POST {path} → {resp.status_code}: {_short_body(resp)}",
+                        status=resp.status_code,
+                    )
+                async for line in resp.aiter_lines():
+                    event = _sse_event(line)
+                    if event is not None:
+                        yield event
+        except CloudError:
+            raise
+        except Exception as e:  # transport/DNS/timeout — no HTTP status
+            raise CloudError(f"POST {path} stream failed: {e}") from e
+
     async def call(
         self,
         method: str,
@@ -185,7 +231,7 @@ class HanzoCloud:
         """Any-method call, for callers driving routes from the OpenAPI spec.
 
         The spec names methods this client has no bespoke helper for (PUT,
-        PATCH, DELETE), so it needs one generic seam rather than a helper per
+        PATCH, DELETE), so it needs one generic entry rather than a helper per
         verb — the auth header and error mapping stay in this one place.
         """
         kw: dict[str, Any] = {}
@@ -201,6 +247,26 @@ class HanzoCloud:
             self._client = None
 
 
+def _sse_event(line: str) -> dict[str, Any] | None:
+    """The event carried by one SSE line, or None when it carries none.
+
+    Blank separators, keep-alive comments, ``event:``/``id:`` fields and the
+    terminal ``[DONE]`` sentinel are all "no event". A ``data:`` payload that is
+    not a JSON object is skipped too: one malformed frame must not end an answer
+    that is still arriving.
+    """
+    if not line.startswith("data:"):
+        return None
+    payload = line[len("data:") :].strip()
+    if not payload or payload == "[DONE]":
+        return None
+    try:
+        event = json.loads(payload)
+    except ValueError:
+        return None
+    return event if isinstance(event, dict) else None
+
+
 def _short_body(resp: Any, limit: int = 300) -> str:
     """A trimmed, single-line error body for CloudError messages."""
     try:

pkg/hanzo-tools/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hanzo-tools"
-version = "0.3.4"
+version = "0.3.5"
 description = "Hanzo AI tools - core infrastructure and tool bundles"
 readme = "README.md"
 requires-python = ">=3.12"