Skip to content
Novarch SDK

Claude Agent SDK

There are two ways to wire the Claude Agent SDK (“ADK”) into Novarch, and which one you want depends on how your agent runs.

  • Agents run by the ADK runtime (query() or ClaudeSDKClient) use the native seam: register one NovarchHookGate and every tool call is gated — your SDK-MCP tools, other MCP servers, and the CLI’s own built-in tools. This is the recommended path.
  • Hand-rolled loops that await SdkMcpTool handlers directly, without the ADK runtime, have no hook seam — they use the decorator floor (augur_async_tool). Still fully supported.

Install the optional extra either way:

Terminal window
pip install "novarch[adk]"

claude_agent_sdk is imported lazily, so import novarch.integrations.claude_agent_sdk works on a bare install.

__all__ = ["NovarchHookGate", "DEFAULT_UNMAPPED_ALLOWLIST", "WriteSpec", "augur_async_tool"]

Declare which tool names are reads and which are gated writes, then hand the gate’s hook configuration to ClaudeAgentOptions:

import asyncio
from claude_agent_sdk import ClaudeAgentOptions, create_sdk_mcp_server, query
from novarch import augur
from novarch.integrations.claude_agent_sdk import NovarchHookGate, WriteSpec
from novarch.schema import GenericContext


@augur.session(tenant_id="acme", team_id="cs-ops", agent_id="cs-bot")
def run_refund_agent(prompt: str) -> None:
    async def _go():
        async with NovarchHookGate(
            tool_kinds={
                "mcp__cs__lookup_customer": "read",
                "mcp__cs__issue_refund": "write",
            },
            write_specs={"mcp__cs__issue_refund": WriteSpec(
                action_type="issue_refund",
                context_builder=lambda args: GenericContext(
                    record_id=args["customer_id"],
                    summary=f"refund {args['customer_id']}",
                ),
            )},
        ) as gate:
            options = ClaudeAgentOptions(
                mcp_servers={"cs": server},
                hooks=gate.hooks_config(),
            )
            async for _ in query(prompt=prompt, options=options):
                pass
            gate.raise_if_killed()   # see "How KILL works here"
    asyncio.run(_go())

Tool names are the names the runtime uses, so MCP tools appear as mcp__<server>__<tool>. The hook fires upstream of allowed_tools — a tool on your allowlist is still gated.

Reads execute and are recorded as evidence when they complete; a write materializes an Action from its WriteSpec and gates on the verdict. Pass the agent’s rationale as a reasoning field in the write tool’s input schema.

How KILL works here

This seam behaves differently from LangChain and LangGraph, and the difference is worth understanding before you rely on it.

The ADK runtime ignores an exception raised inside a hook — it logs the error and runs the tool anyway — and its “stop the run” flag does not halt a run either. So the gate never raises from a hook. Instead:

  • A blocked write is refused at the hook, and the agent is told why.
  • The gate then latches: every later tool call in that session, reads included, is refused without evaluation.
  • gate.raise_if_killed() after your query loop re-raises the NovarchKillError your except blocks expect.

Call raise_if_killed() — without it, a killed session ends quietly and your calling code never learns the run was stopped. The enforcement itself does not depend on it (nothing gated executes after the latch), but your error handling does.

Verdict mapping

VerdictWhat the agent sees
judge pass / operator approvethe tool runs
judge killthe call is refused, the session latches, raise_if_killed() raises NovarchKillError
operator denythe call is refused with the rationale; the agent continues
operator timeout / skiprefused + latched (fail-closed)

On a pass the gate returns no opinion rather than an approval, so your own permission rules still apply — Novarch narrows what an agent may do, never widens it.

Unmapped tools

A tool missing from tool_kinds is refused: unmapped tools never execute ungated. One name is allowlisted by default —

DEFAULT_UNMAPPED_ALLOWLIST = frozenset({"ToolSearch"})

ToolSearch is how the runtime loads MCP tool schemas; refusing it stops the agent before it can reach any gated tool at all. Allowlisted calls are still recorded as evidence. Widen it if your agent needs other built-ins:

NovarchHookGate(..., allow_unmapped={"ToolSearch", "Read"})

Hold timeouts

hooks_config() sets the hook timeout to the wait this session negotiated plus a minute of headroom. The runtime’s own default of 60 seconds is far below a realistic operator hold, so take the derived value:

gate.hooks_config()

A hook that errors on this seam fails open. A matcher that expires before the gate’s own fail-close would release a write no judge and no operator ever saw, so hooks_config() refuses a timeout_s that does not strictly outlive the session’s wait, and raises ValueError naming the number it needs.

A fixed override goes stale. The wait is negotiated at session entry against the deployment’s hold window, so a number written when a hold could only last 300 seconds is too short the moment an admin widens that window or a session opts in with max_hold_s=augur.ADOPT_DEPLOYMENT_WINDOW. Set one only if you have to, and set it above the largest hold window your deployment allows.

Parallel tool calls

The runtime dispatches hook callbacks concurrently, and the gate is built for that: parallel reads are fully supported, while a write dispatched alongside other gated calls is fail-closed refused, so a write is never judged on partially-collected evidence and never slips past a sibling’s kill.

One limit worth knowing: a read whose hook has not fired yet is invisible to a write that has already been checked. The hook payload carries no view of the other tool calls in the turn, so a write scheduled ahead of its turn’s reads is judged on the evidence collected so far. Keeping a write in its own turn — the pattern the gate already pushes agents toward — avoids this.

augur_async_tool (hand-rolled loops)

If you await SdkMcpTool handlers yourself instead of running the ADK runtime, there is no hook seam — use the decorator floor.

ADK tools are async, and its @tool returns an SdkMcpTool dataclass rather than a plain callable, so the wrapping has to happen at the ADK boundary. The stacking order is ADK-outside / augur-inside: ADK’s @tool checks inspect.iscoroutinefunction on the handler, so the augur layer must run first and produce an async wrapper (which @augur.tool does for async def bodies). augur_async_tool does both in one decorator and a single import, then returns the SdkMcpTool that create_sdk_mcp_server expects.

def augur_async_tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any],
    *,
    kind: Literal["read", "write"],
    action_type: str | None = None,
    context_builder: Callable[..., ActionContext] | None = None,
) -> Callable[[Callable[..., Awaitable[dict[str, Any]]]], Any]: ...

The wrapped function must be async def. Augur’s own validation applies: kind="write" requires action_type and context_builder.

from novarch.integrations.claude_agent_sdk import augur_async_tool
from novarch.schema import GenericContext


@augur_async_tool(
    "issue_refund",
    "Issue a refund",
    {"customer_id": str, "amount_cents": int},
    kind="write",
    action_type="issue_refund",
    context_builder=lambda args: GenericContext(
        record_id=args["customer_id"],
        summary=f"refund {args['customer_id']}",
        amount=args["amount_cents"] / 100,
    ),
)
async def issue_refund(args):
    return {"content": [{"type": "text", "text": "refund ok"}]}


server = create_sdk_mcp_server(name="cs", tools=[issue_refund])

On the decorator path a kill or operator-deny surfaces as NovarchKillError raised from the tool body directly — no latch, no raise_if_killed() — because your own code awaits the handler.

@augur.session is polymorphic: an async def entry point gets an async wrapper. The common pattern is a sync session wrapping the drive in asyncio.run; ContextVars copy into the event loop at task creation, so the active session is reachable from every tool handler.

See augur decorators for the underlying @augur.tool semantics and Concepts for the Novarch engine.