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.19.0#subdirectory=agents/sdk"

Replace @v2.19.0 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,               # protects from auto-archive
            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 run_cycle(self, client: GovernanceClient) -> CycleResult | None:
        # Do your work here. Return a CycleResult to trigger a check-in,
        # or None to skip (useful for "nothing to do this tick" paths).
        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))

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).

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.

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.