Client

agent_assembly.client.GatewayClient is the HTTP client that talks to the governance gateway. It is created by init_assembly() and exposed as AssemblyContext.client for cases where user code needs to issue ad-hoc gateway calls outside the framework adapter path.

agent_assembly.client.GatewayClient

agent_assembly.client.GatewayClient

GatewayClient(
    gateway_url: str,
    agent_id: str,
    api_key: str | None = None,
    timeout: int = 30,
    *,
    control_plane_url: str | None = None,
    parent_agent_id: str | None = None,
    team_id: str | None = None,
    delegation_reason: str | None = None,
    spawned_by_tool: str | None = None,
    depth: int | None = None,
    enforcement_mode: str | None = None,
    allow_insecure: bool = False,
)

Client for communicating with the Agent Assembly governance gateway.

PARAMETER	DESCRIPTION
gateway_url	URL of the governance gateway TYPE: str
agent_id	Unique identifier for the agent TYPE: str
api_key	Optional API key for authentication TYPE: str | None DEFAULT: None
timeout	Request timeout in seconds TYPE: int DEFAULT: 30
control_plane_url	Optional URL of the control-plane HTTP API. When set, the remaining HTTP routes (/topology/edges, /dispatch_tool) are issued against it; when None (the default) they fall back to gateway_url — the backwards-compatible single-host OSS dev setup. Agent registration and policy checks go through the native gRPC path (ADR 0004), not these HTTP routes. TYPE: str | None DEFAULT: None
parent_agent_id	Parent agent ID for topology tracking TYPE: str | None DEFAULT: None
team_id	Team ID this agent belongs to TYPE: str | None DEFAULT: None
delegation_reason	Human-readable reason for delegation TYPE: str | None DEFAULT: None
spawned_by_tool	Name of the tool that spawned this agent TYPE: str | None DEFAULT: None
depth	Spawn depth in the agent lineage tree TYPE: int | None DEFAULT: None
enforcement_mode	Per-agent governance posture for this agent. None (the default) lets the gateway apply its server-side default of live enforcement. Stored for reference; registration itself now goes through the native gRPC path (ADR 0004). TYPE: str | None DEFAULT: None
allow_insecure	Permit sending the Authorization: Bearer API key over plaintext http:// to a non-loopback host. Off by default — a remote http:// base URL with an API key set is refused with :class:ValueError, because the Bearer credential would travel unencrypted (AAASM-3725). Loopback http:// is always allowed (local dev gateway). TYPE: bool DEFAULT: False

Source code in agent_assembly/client/gateway.py

def __init__(
    self,
    gateway_url: str,
    agent_id: str,
    api_key: str | None = None,
    timeout: int = 30,
    *,
    control_plane_url: str | None = None,
    parent_agent_id: str | None = None,
    team_id: str | None = None,
    delegation_reason: str | None = None,
    spawned_by_tool: str | None = None,
    depth: int | None = None,
    enforcement_mode: str | None = None,
    allow_insecure: bool = False,
) -> None:
    """
    Initialize the GatewayClient.

    Args:
        gateway_url: URL of the governance gateway
        agent_id: Unique identifier for the agent
        api_key: Optional API key for authentication
        timeout: Request timeout in seconds
        control_plane_url: Optional URL of the control-plane HTTP API. When
            set, the remaining HTTP routes (``/topology/edges``,
            ``/dispatch_tool``) are issued against it; when ``None`` (the
            default) they fall back to ``gateway_url`` — the backwards-compatible
            single-host OSS dev setup. Agent registration and policy checks
            go through the native gRPC path (ADR 0004), not these HTTP routes.
        parent_agent_id: Parent agent ID for topology tracking
        team_id: Team ID this agent belongs to
        delegation_reason: Human-readable reason for delegation
        spawned_by_tool: Name of the tool that spawned this agent
        depth: Spawn depth in the agent lineage tree
        enforcement_mode: Per-agent governance posture for this agent.
            ``None`` (the default) lets the gateway apply its server-side
            default of live enforcement. Stored for reference; registration
            itself now goes through the native gRPC path (ADR 0004).
        allow_insecure: Permit sending the ``Authorization: Bearer`` API key
            over plaintext ``http://`` to a **non-loopback** host. Off by
            default — a remote ``http://`` base URL with an API key set is
            refused with :class:`ValueError`, because the Bearer credential
            would travel unencrypted (AAASM-3725). Loopback ``http://`` is
            always allowed (local dev gateway).
    """
    self.gateway_url = gateway_url.rstrip("/")
    self.control_plane_url = control_plane_url.rstrip("/") if control_plane_url else None
    self.agent_id = agent_id
    self.api_key = api_key
    self.timeout = timeout
    self.parent_agent_id = parent_agent_id
    self.team_id = team_id
    self.delegation_reason = delegation_reason
    self.spawned_by_tool = spawned_by_tool
    self.depth = depth
    self.enforcement_mode = enforcement_mode
    self.allow_insecure = allow_insecure
    self._client: httpx.Client | None = None

client property

client: Client

Get or create the HTTP client.

The base URL is control_plane_url when one was supplied, otherwise it falls back to gateway_url (single-host OSS dev).

close

close() -> None

Close the HTTP client.

Source code in agent_assembly/client/gateway.py

def close(self) -> None:
    """Close the HTTP client."""
    if self._client is not None:
        self._client.close()
        self._client = None

report_edge

report_edge(
    source_agent_id: str,
    target_agent_id: str,
    edge_type: str,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]

Report a directed edge between two agents to the topology store.

PARAMETER	DESCRIPTION
source_agent_id	ID of the source agent TYPE: str
target_agent_id	ID of the target agent TYPE: str
edge_type	Semantic type of the edge (e.g. "messages", "delegates_to") TYPE: str
metadata	Optional freeform metadata to attach to the edge TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	Response data containing the assigned edge id

RAISES	DESCRIPTION
GatewayError	If the request fails

Source code in agent_assembly/client/gateway.py

def report_edge(
    self,
    source_agent_id: str,
    target_agent_id: str,
    edge_type: str,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """
    Report a directed edge between two agents to the topology store.

    Args:
        source_agent_id: ID of the source agent
        target_agent_id: ID of the target agent
        edge_type: Semantic type of the edge (e.g. "messages", "delegates_to")
        metadata: Optional freeform metadata to attach to the edge

    Returns:
        Response data containing the assigned edge id

    Raises:
        GatewayError: If the request fails
    """
    import json as _json

    body: dict[str, str] = {
        "source_agent_id": source_agent_id,
        "target_agent_id": target_agent_id,
        "edge_type": edge_type,
    }
    if metadata is not None:
        body["metadata_json"] = _json.dumps(metadata)
    try:
        response = self.client.post("/topology/edges", json=body)
        response.raise_for_status()
        data: dict[str, Any] = response.json()
        return data
    except httpx.HTTPError as e:
        raise GatewayError(f"Failed to report edge: {e}") from e

dispatch_tool async

dispatch_tool(
    tool_name: str, args: dict[str, Any]
) -> DispatchToolResult

Dispatch a tool with placeholder-form args (AAASM-1920 Secret Injection).

The gateway resolves every ${NAME} placeholder in args against its registered SecretsStore, emits a placeholder-form audit entry, and returns the resolved args plus the list of substituted names.

The LLM never observes the resolved credential value: the agent code holds the placeholder, Assembly resolves it on the gateway side, and the response is forwarded to the tool sink only.

PARAMETER	DESCRIPTION
tool_name	Name of the tool to dispatch (e.g. "call_database"). TYPE: str
args	Placeholder-form args. May contain ${NAME} tokens that will be resolved server-side before the tool is invoked. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
DispatchToolResult	DispatchToolResult with the resolved args + the list of
DispatchToolResult	placeholder names that were substituted.

RAISES	DESCRIPTION
GatewayError	If the request fails for any reason — including a 422 Unprocessable Entity when args references a placeholder that is not registered in the gateway's SecretsStore.

Source code in agent_assembly/client/gateway.py

async def dispatch_tool(self, tool_name: str, args: dict[str, Any]) -> DispatchToolResult:
    """
    Dispatch a tool with placeholder-form args (AAASM-1920 Secret Injection).

    The gateway resolves every ``${NAME}`` placeholder in ``args`` against
    its registered ``SecretsStore``, emits a placeholder-form audit entry,
    and returns the resolved args plus the list of substituted names.

    The LLM never observes the resolved credential value: the agent code
    holds the placeholder, Assembly resolves it on the gateway side, and
    the response is forwarded to the tool sink only.

    Args:
        tool_name: Name of the tool to dispatch (e.g. ``"call_database"``).
        args: Placeholder-form args. May contain ``${NAME}`` tokens that
            will be resolved server-side before the tool is invoked.

    Returns:
        ``DispatchToolResult`` with the resolved args + the list of
        placeholder names that were substituted.

    Raises:
        GatewayError: If the request fails for any reason — including a
            422 Unprocessable Entity when ``args`` references a placeholder
            that is not registered in the gateway's ``SecretsStore``.
    """
    body = {"tool": tool_name, "args": args}
    try:
        response = self.client.post("/dispatch_tool", json=body)
        response.raise_for_status()
        data = response.json()
        return DispatchToolResult(
            resolved_args=data.get("resolved_args", {}),
            names_substituted=list(data.get("names_substituted", [])),
        )
    except httpx.HTTPError as e:
        raise GatewayError(f"Failed to dispatch tool {tool_name}: {e}") from e