mirror of
https://github.com/hanzoai/agent.git
synced 2026-09-11 21:08:51 +00:00
network: wire cross-process agent dispatch over ZAP, gated on HANZO_AGENT_WIRE=zap
Closes the Hanzo-side LP-201 Tier 1 control-plane migration step. ZapClient
+ ZapAgent + consensus primitives already shipped in src/agents/zap/; this
landing wires the AgentNetwork-side cross-process lookup path.
New: agents/network/remote.py
- RemoteAgentProxy: value-shaped facade quacking like Agent on the routing
side (name + handoff_description) and dispatching invocations through
ZapClient.call_tool(`agent/<name>`, …) on the wire side.
- is_zap_wire_enabled(): single source of truth for the feature flag —
one function, one place (Hickey-style profile gate). Reads HANZO_AGENT_WIRE
case-insensitively; only the literal "zap" enables. Anything else
fail-closed (json, "true", "1", unset → JSON/local-agent path).
AgentNetwork.add_remote_agent(name, zap_uri, …): cross-process counterpart
of add_agent. Holds a NetworkNode pointing at a RemoteAgentProxy; the router
reads it like a local agent (same capabilities/metadata surface). At
dispatch the proxy's invoke() pushes over the ZAP wire — and raises with a
clear message if the operator forgot to set the env var.
Default path (no env) keeps the JSON/OpenAI-compatible wire — external API
consumers see no change. ZAP is intra-cluster only.
Tests: 7 unit tests covering both feature-flag combinations + the clone()
mirror + the fail-loud guard, all green via --noconftest (existing conftest
pulls heavy deps; tests load remote.py directly via importlib, mirroring
test_zap.py).
This commit is contained in:
@@ -19,6 +19,7 @@ from ..run_context import RunContextWrapper, TContext
|
||||
from ..state.store import StateStore, InMemoryStateStore
|
||||
from ..tracing import custom_span, get_current_trace
|
||||
from .node import NetworkNode, NodeStatus
|
||||
from .remote import RemoteAgentProxy, is_zap_wire_enabled
|
||||
from .router import Router, RoutingDecision, SemanticRouter
|
||||
|
||||
|
||||
@@ -116,6 +117,68 @@ class AgentNetwork:
|
||||
|
||||
return node
|
||||
|
||||
def add_remote_agent(
|
||||
self,
|
||||
name: str,
|
||||
zap_uri: str,
|
||||
*,
|
||||
handoff_description: str | None = None,
|
||||
capabilities: List[str] | None = None,
|
||||
dependencies: List[str] | None = None,
|
||||
metadata: Dict[str, Any] | None = None,
|
||||
) -> NetworkNode:
|
||||
"""Add a *remote* agent that lives at the other end of a ZAP wire.
|
||||
|
||||
The local network holds only a `RemoteAgentProxy`; invocations are
|
||||
dispatched over ZAP to the peer endpoint, which is expected to expose
|
||||
the named agent as a ZAP tool (`agent/<name>`). This is the
|
||||
cross-process counterpart of `add_agent` and is gated on the
|
||||
`HANZO_AGENT_WIRE=zap` env var at invocation time. Registering a
|
||||
remote agent without the env set is legal — it errors at dispatch.
|
||||
|
||||
Args:
|
||||
name: Remote agent name (matches the peer-side registration).
|
||||
zap_uri: ZAP endpoint URI for the peer.
|
||||
handoff_description: Description used when the agent is handed off.
|
||||
capabilities: Router capabilities advertised by the remote peer.
|
||||
dependencies: Dependency names this remote depends on.
|
||||
metadata: Additional routing metadata.
|
||||
|
||||
Returns:
|
||||
The network node wrapping the remote agent proxy.
|
||||
"""
|
||||
if name in self.nodes:
|
||||
raise AgentsException(f"Agent '{name}' already exists in network")
|
||||
|
||||
proxy = RemoteAgentProxy(
|
||||
name=name,
|
||||
zap_uri=zap_uri,
|
||||
handoff_description=handoff_description,
|
||||
capabilities=list(capabilities or []),
|
||||
metadata=dict(metadata or {}),
|
||||
)
|
||||
|
||||
# NetworkNode is typed `Agent`-shaped; RemoteAgentProxy carries the
|
||||
# same minimal public surface (name + handoff_description + handoffs),
|
||||
# so it satisfies the structural needs of the router and dispatcher.
|
||||
# We `cast` rather than widen NetworkNode's type to keep the local-
|
||||
# agent path unchanged.
|
||||
node = NetworkNode(
|
||||
agent=cast(Agent, proxy),
|
||||
capabilities=list(capabilities or []),
|
||||
dependencies=list(dependencies or []),
|
||||
metadata=dict(metadata or {}),
|
||||
)
|
||||
|
||||
self.nodes[name] = node
|
||||
self.router.update_agent_info(name, node.capabilities, node.metadata)
|
||||
|
||||
logger.debug(
|
||||
"Added remote agent '%s' via ZAP at %s (capabilities: %s)",
|
||||
name, zap_uri, node.capabilities,
|
||||
)
|
||||
return node
|
||||
|
||||
def remove_agent(self, agent_name: str) -> None:
|
||||
"""Remove an agent from the network."""
|
||||
if agent_name not in self.nodes:
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Cross-process agent invocation over ZAP wire.
|
||||
|
||||
When `HANZO_AGENT_WIRE=zap` is set, an `AgentNetwork` can hold *remote* nodes
|
||||
backed by a `ZapClient` instead of a local `Agent`. Calls to a remote node
|
||||
are dispatched as ZAP `call_tool` requests at the peer endpoint named
|
||||
`agent/<name>`; the peer side is expected to expose its agents as ZAP tools.
|
||||
|
||||
Default wire stays JSON/OpenAI-compatible. ZAP is intra-cluster only — it is
|
||||
opt-in via env-var so external API consumers see no change.
|
||||
|
||||
Decomplected from `AgentNetwork`:
|
||||
- Network: owns the nodes dict + the router + dispatch.
|
||||
- RemoteAgentProxy: a value-shaped facade that quacks like `Agent` enough
|
||||
for routing/dispatch (name + handoff_description) and that, when invoked,
|
||||
pushes the payload over ZAP rather than driving a local LLM loop.
|
||||
- is_zap_wire_enabled: a single function in a single place that reads the
|
||||
feature flag. Profile gate, Hickey-style.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional
|
||||
|
||||
# Soft import — ZapClient lives in agents/zap; importing here would create a
|
||||
# cycle at module load time if zap pulled in network. So we import lazily
|
||||
# from inside connect().
|
||||
|
||||
|
||||
def is_zap_wire_enabled() -> bool:
|
||||
"""Single source of truth for the ZAP cross-process feature flag.
|
||||
|
||||
Read once per process via `os.environ`. The default (no env, or any value
|
||||
other than the literal string ``zap``) keeps the network on the JSON/local
|
||||
Agent path — preserving OpenAI compatibility for external API consumers.
|
||||
"""
|
||||
return os.environ.get("HANZO_AGENT_WIRE", "").strip().lower() == "zap"
|
||||
|
||||
|
||||
@dataclass
|
||||
class RemoteAgentProxy:
|
||||
"""A value-shaped proxy for an agent that lives at the other end of a ZAP wire.
|
||||
|
||||
Shapes the public surface of `Agent` minimally — enough for the network
|
||||
router (which only reads `name` + `handoff_description`) and for the
|
||||
dispatcher (which calls `invoke` rather than driving the OpenAI runner).
|
||||
"""
|
||||
|
||||
name: str
|
||||
"""The remote agent's name as it is registered on the peer side."""
|
||||
|
||||
zap_uri: str
|
||||
"""ZAP endpoint URI for the peer (e.g. ``zap://peer-2.cluster:9999``)."""
|
||||
|
||||
handoff_description: Optional[str] = None
|
||||
"""Human description used when the agent is exposed as a handoff."""
|
||||
|
||||
capabilities: list[str] = field(default_factory=list)
|
||||
"""Routing capabilities advertised by the remote peer."""
|
||||
|
||||
metadata: dict[str, Any] = field(default_factory=dict)
|
||||
"""Additional routing metadata (model, region, etc.)."""
|
||||
|
||||
# The Agent dataclass contract carries these fields as well; we keep them
|
||||
# default-empty so any code that reads them via getattr does not break.
|
||||
handoffs: list[Any] = field(default_factory=list)
|
||||
tools: list[Any] = field(default_factory=list)
|
||||
instructions: Optional[str] = None
|
||||
|
||||
async def invoke(self, input: Any, *, context: Any = None) -> Any:
|
||||
"""Dispatch an invocation to the remote agent over ZAP.
|
||||
|
||||
The peer side exposes each registered agent as a ZAP tool named
|
||||
``agent/<name>`` taking a single ``input`` argument and returning the
|
||||
agent's response payload.
|
||||
"""
|
||||
if not is_zap_wire_enabled():
|
||||
raise RuntimeError(
|
||||
"RemoteAgentProxy.invoke called with HANZO_AGENT_WIRE != zap"
|
||||
)
|
||||
|
||||
# Lazy import to avoid network <-> zap module cycle on package init.
|
||||
from ..zap.client import ZapClient
|
||||
|
||||
async with await ZapClient.from_uri(self.zap_uri) as client:
|
||||
return await client.call_tool(
|
||||
f"agent/{self.name}",
|
||||
{"input": input, "context": context},
|
||||
)
|
||||
|
||||
def clone(self, **kwargs: Any) -> "RemoteAgentProxy":
|
||||
"""Mirrors Agent.clone — produce a derived proxy with overrides."""
|
||||
return RemoteAgentProxy(
|
||||
name=kwargs.get("name", self.name),
|
||||
zap_uri=kwargs.get("zap_uri", self.zap_uri),
|
||||
handoff_description=kwargs.get(
|
||||
"handoff_description", self.handoff_description
|
||||
),
|
||||
capabilities=list(kwargs.get("capabilities", self.capabilities)),
|
||||
metadata=dict(kwargs.get("metadata", self.metadata)),
|
||||
handoffs=list(kwargs.get("handoffs", self.handoffs)),
|
||||
tools=list(kwargs.get("tools", self.tools)),
|
||||
instructions=kwargs.get("instructions", self.instructions),
|
||||
)
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Tests for the cross-process ZAP agent proxy (network/remote.py).
|
||||
|
||||
Loads `remote.py` directly via importlib to avoid the heavyweight
|
||||
`agents.__init__` import chain (numpy / griffe transitive deps), mirroring
|
||||
the pattern used in `test_zap.py`.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def _load_module(name: str, path: str):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
mod = importlib.util.module_from_spec(spec) # type: ignore
|
||||
sys.modules[name] = mod
|
||||
spec.loader.exec_module(mod) # type: ignore
|
||||
return mod
|
||||
|
||||
|
||||
_here = os.path.dirname(os.path.abspath(__file__))
|
||||
_base = os.path.join(_here, "..", "src", "agents", "network")
|
||||
_remote = _load_module("agents.network.remote", os.path.join(_base, "remote.py"))
|
||||
|
||||
|
||||
RemoteAgentProxy = _remote.RemoteAgentProxy
|
||||
is_zap_wire_enabled = _remote.is_zap_wire_enabled
|
||||
|
||||
|
||||
def test_feature_flag_default_disabled(monkeypatch):
|
||||
"""Default (no env) keeps the JSON/local-agent path — OpenAI-compatible."""
|
||||
monkeypatch.delenv("HANZO_AGENT_WIRE", raising=False)
|
||||
assert is_zap_wire_enabled() is False
|
||||
|
||||
|
||||
def test_feature_flag_enabled_with_zap(monkeypatch):
|
||||
"""`HANZO_AGENT_WIRE=zap` enables the cross-process wire."""
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "zap")
|
||||
assert is_zap_wire_enabled() is True
|
||||
|
||||
|
||||
def test_feature_flag_case_insensitive(monkeypatch):
|
||||
"""Uppercase / mixed-case values still flip the flag (operator friendliness)."""
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "ZAP")
|
||||
assert is_zap_wire_enabled() is True
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "Zap")
|
||||
assert is_zap_wire_enabled() is True
|
||||
|
||||
|
||||
def test_feature_flag_other_values_disabled(monkeypatch):
|
||||
"""Anything other than literal 'zap' keeps the flag off — fail-closed default."""
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "json")
|
||||
assert is_zap_wire_enabled() is False
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "1")
|
||||
assert is_zap_wire_enabled() is False
|
||||
monkeypatch.setenv("HANZO_AGENT_WIRE", "true")
|
||||
assert is_zap_wire_enabled() is False
|
||||
|
||||
|
||||
def test_proxy_construction_defaults():
|
||||
"""Defaults: empty capabilities/metadata/handoffs/tools, no instructions."""
|
||||
p = RemoteAgentProxy(name="peer1", zap_uri="zap://peer-1.cluster:9999")
|
||||
assert p.name == "peer1"
|
||||
assert p.zap_uri == "zap://peer-1.cluster:9999"
|
||||
assert p.handoff_description is None
|
||||
assert p.capabilities == []
|
||||
assert p.metadata == {}
|
||||
assert p.handoffs == []
|
||||
assert p.tools == []
|
||||
assert p.instructions is None
|
||||
|
||||
|
||||
def test_proxy_clone_preserves_and_overrides():
|
||||
"""clone() mirrors Agent.clone — fields passed are overridden; rest preserved."""
|
||||
base = RemoteAgentProxy(
|
||||
name="peer1",
|
||||
zap_uri="zap://peer-1.cluster:9999",
|
||||
handoff_description="code review agent",
|
||||
capabilities=["code", "review"],
|
||||
metadata={"region": "us-west"},
|
||||
)
|
||||
cloned = base.clone(name="peer1-shadow")
|
||||
assert cloned.name == "peer1-shadow"
|
||||
assert cloned.zap_uri == base.zap_uri
|
||||
assert cloned.handoff_description == base.handoff_description
|
||||
assert cloned.capabilities == base.capabilities
|
||||
assert cloned.metadata == base.metadata
|
||||
# And confirm the override is shallow-copied (mutating the clone doesn't
|
||||
# bleed into the original — caps lists are independent).
|
||||
cloned.capabilities.append("ops")
|
||||
assert base.capabilities == ["code", "review"]
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invoke_errors_without_feature_flag(monkeypatch):
|
||||
"""invoke() refuses to dispatch when the wire feature-flag is off.
|
||||
|
||||
Fail-loud: avoids silent regression to a misconfigured local-dispatch
|
||||
path when the operator forgets to set HANZO_AGENT_WIRE=zap on a node
|
||||
that holds remote proxies.
|
||||
"""
|
||||
monkeypatch.delenv("HANZO_AGENT_WIRE", raising=False)
|
||||
p = RemoteAgentProxy(name="peer1", zap_uri="zap://peer-1.cluster:9999")
|
||||
with pytest.raises(RuntimeError, match="HANZO_AGENT_WIRE"):
|
||||
await p.invoke("hello")
|
||||
Reference in New Issue
Block a user