unitares-sdk

Build your own Unitares resident agent. A resident is a long-running (or scheduled) process that checks in to governance, carries an EISV state vector, and participates in the shared knowledge graph. Vigil, Sentinel, and Chronicler are reference implementations.

Install

pip install unitares-sdk

To install the SDK as it shipped with a specific server release, use the Git URL instead:

pip install "unitares-sdk @ git+https://github.com/cirwel/unitares@v2.22.1#subdirectory=agents/sdk"

Replace @v2.22.1 with another server release tag only after checking the compatibility map. Pin a commit SHA when reproducing a specific development build.

Or from a checkout of the unitares repo:

pip install -e agents/sdk

Releases are tag-driven. Pushing an sdk-v* tag whose version matches version in pyproject.toml runs .github/workflows/publish-sdk.yml, which builds and uploads via PyPI Trusted Publishing — OIDC, no API token. The SDK carries its own version series, independent of the server's. See the release process.

Links in this file are absolute on purpose: pyproject.toml uses it as the package readme, so PyPI renders it outside the repo and relative paths 404.

The package is standalone — it talks to a UNITARES server over MCP/REST and never needs the server codebase importable. Fully typed (py.typed).

The 30-line resident

from pathlib import Path
from unitares_sdk.agent import CycleResult, GovernanceAgent
from unitares_sdk.client import GovernanceClient


class MyResident(GovernanceAgent):
    def __init__(self):
        super().__init__(
            name="MyResident",
            mcp_url="http://127.0.0.1:8767/mcp/",
            persistent=True,               # requires roster registration — see below
            refuse_fresh_onboard=True,     # explicit bootstrap required
            cycle_timeout_seconds=60.0,    # hard cap on one cycle
            log_file=Path("/tmp/my_resident.log"),
            max_log_lines=10_000,
        )

    async def do_scan(self, client: GovernanceClient) -> int:
        # Your work goes here. Returning 0 means "nothing to do this tick".
        return 1

    async def run_cycle(self, client: GovernanceClient) -> CycleResult | None:
        # Return a CycleResult to trigger a check-in, or None to skip.
        count = await self.do_scan(client)
        if count == 0:
            return None
        return CycleResult(
            summary=f"scanned {count} items",
            complexity=0.2,
            confidence=0.9,
        )


if __name__ == "__main__":
    import asyncio
    asyncio.run(MyResident().run_forever(interval=60))

Register the name first

Before the first run, add the agent's name to the governance server's UNITARES_RESIDENTS roster and restart the server:

UNITARES_RESIDENTS=MyResident            # comma-separated; empty by default

This is not optional bookkeeping. persistent and autonomous are privileged tags: the server grants them only at mint, and only when the name being minted is on that roster. An identity cannot assign them to itself afterwards, so a resident bootstrapped under an unlisted name is not protected from auto-archive and cannot be upgraded in place — the only fix is to register the name and bootstrap a fresh identity.

The roster is empty by default and never ships a fleet, so a fresh install has no named residents until you declare yours.

First run

UNITARES_FIRST_RUN=1 python my_resident.py — this mints the identity and stores its UUID anchor at ~/.unitares/anchors/myresident.json. Every subsequent run resumes that anchor automatically. Never delete anchors: if you do, set UNITARES_FIRST_RUN=1 again to re-bootstrap (you will get a new UUID).

If the name was not on the roster, the mint raises ResidentRegistrationRefused with the exact remedy rather than starting an agent that quietly is not a resident. The server reports what happened in OnboardResult.resident_registration:

{"status": "not_on_roster",       # or: registered | no_roster_configured
 "requested_name": "MyResident",
 "granted_tags": ["ephemeral"],
 "required_tags": ["persistent", "autonomous"],
 "roster_env": "UNITARES_RESIDENTS",
 "detail": "..."}                 # actionable remedy

Pass persistent=False for an ordinary long-running agent that does not need resident semantics; no roster entry is required for that.

Extension points

The base class handles MCP connect, identity resolve, check-in, heartbeat, log rotation, state persistence, and graceful shutdown. Override these to extend behavior:

Hook exceptions are logged and swallowed — a broken hook cannot take down a cycle. asyncio.CancelledError always propagates.

Do not override _ensure_identity, _handle_cycle_result, or _send_heartbeat — those are load-bearing and change across versions.

Constructor reference

Lifecycle shapes

• Daemon: asyncio.run(agent.run_forever(interval=60)) — loops forever with heartbeats when idle. Reference: agents/sentinel/agent.py.
• Scheduled: asyncio.run(agent.run_once()) under launchd / systemd cron. References: agents/chronicler/agent.py, agents/vigil/agent.py.

Both shapes respect cycle_timeout_seconds and auto-trim log_file.

State persistence

• self.save_state(d: dict) and self.load_state() -> dict let your run_cycle carry data across iterations. Writes are atomic (os.replace).
• Non-JSON-serializable values (e.g. datetime, Path) are coerced to their str() representation on save rather than raising TypeError. You get a lossy round-trip, not a silent state-write failure.

Identity rules

1. The agent's first MCP call (onboard or identity) is the sole source of identity. Do not set identity out-of-band.
2. UUID is the ground truth. client_session_id and continuity_token are cache keys for ephemeral clients; residents don't need them.
3. Anchors live at ~/.unitares/anchors/<name_lower>.json. One anchor per host per role. The file contains agent_uuid and is written atomically.
4. Never silent-swap an identity. If the anchor is missing and refuse_fresh_onboard=True, _ensure_identity raises IdentityBootstrapRefused — the operator must explicitly set UNITARES_FIRST_RUN=1 once to mint a new one.

Identity-bound lease calls

LeasePlaneClient accepts identity_proof=<continuity_token> on acquire, renew, heartbeat, release, handoff_offer, and handoff_accept. Before each mutation it exchanges that credential with governance for a short-lived lat.v1 attestation bound to the exact method, path, and serialized request body. Only the attestation travels in X-Unitares-Identity-Proof; neither proof is persisted in the request body. Remote governance exchanges require HTTPS; plain HTTP is accepted only on loopback or for an explicitly listed internal hostname. A caller may pass an already-minted lat.v1 token, but it is single-use: acquire_with_retry requires identity_proof_factory so every attempt receives a fresh token. hybrid mode supports mixed-version upgrades; attestation mode fails closed on raw proofs. The SDK does not forward a continuity credential when exchange fails unless an operator deliberately sets identity_legacy_fallback=True in its client config for that migration window.

Not in the SDK (on purpose)

• agents/common/findings.py and agents/common/config.py are internal to the reference residents in this repo. If you need findings-posting in your own resident, vendor the helper or POST to /api/findings yourself — the REST contract is the public surface.
• The violation taxonomy is server-owned vocabulary (src/violation_taxonomy.yaml); consume it via the /v1/taxonomy endpoint rather than importing server code.
• Watcher (agents/watcher/agent.py) uses a different execution model (sync, hook-driven, one-shot per tool-use event) and does not subclass GovernanceAgent.

NeMo Relay integration

unitares_sdk.integrations.nemo_relay attaches UNITARES to NVIDIA NeMo Relay, the agent runtime that owns lifecycle events and tool-boundary middleware for Python, Node and Rust agents. Install the extra and register the plugin kind:

pip install "unitares-sdk[nemo-relay]"

from nemo_relay import plugin as relay_plugin
from unitares_sdk.integrations.nemo_relay import install

unitares = install()   # registers plugin kind "unitares"
await relay_plugin.initialize(relay_plugin.PluginConfig(components=[
    relay_plugin.ComponentSpec(kind="unitares", config={
        "rest_url": "http://127.0.0.1:8767/v1/tools/call",
        "checkin_every_tool_calls": 20,
    }),
]))

What it does, and what it deliberately does not claim:

• One identity per run. Each root Relay scope mints a fresh UNITARES identity (force_new=true), the strict-identity contract's normal path. Nested agent scopes share it unless subagent_identities is on, in which case each becomes a lineage-linked identity with spawn_reason="subagent".
• Substrate check-ins, not agent reports. Tool and LLM counts are filed every checkin_every_tool_calls tool calls and at run end under epistemic_class="substrate_interpretation", with no confidence, because the integration holds no belief about the work.
• Outcomes with honest provenance. Tool errors become task_failed and guardrail rejections become tool_rejected outcome events. They keep the server's default verification_source; the integration runs inside the agent's process, so this is in-band evidence, not an external signal. Rejections issued by the integration's own gate are never filed as outcomes, so the governor does not grade its own enforcement.
• A contestable gate. With enforce on (the default), a Relay conditional-execution guardrail refuses tool calls for a run whose latest policy action is pause or reject. The refusal names the verdict, the guidance, and the review path. The gate never touches the network, so enforcement latency is bounded by the check-in cadence.
• Fail-open by default. A scope the exporter has not bound yet, or an unreachable server, allows the call and logs. fail_closed: true refuses instead.

Network calls run on a private worker thread; the Relay subscriber only enqueues. Tests inject a fake client through install(client_factory=...), so the integration is exercised without a server; the end-to-end test against the real runtime skips when the extra is absent.