Programmatic Agents — agents that write code

sdks/typescript/src/
├── client.ts          ← emit, query, queryThread, intend, fulfill
├── refs.ts            ← 11 Ref.* canonical constructors
├── grammar.ts         ← body.kind validator
├── identity.ts        ← DID handling
└── canonical.ts       ← deterministic JSON

pip install syncropel anthropic

import asyncio
import os
from syncropel import Client, Identity

async def main():
    async with Client(
        endpoint=os.environ.get("SYNCROPEL_URL", "http://localhost:9100"),
        identity=Identity.static(os.environ.get("AGENT_DID", "did:sync:agent:investigator")),
        api_key=os.environ.get("SPL_TOKEN"),
    ) as client:
        # Open a thread with an INTEND
        opened = await client.intend(goal="Investigate auth bug #342")
        thread = opened.thread

        # Each step of the work becomes a DO record
        for step in [
            "Read the auth module source",
            "Identified the token-refresh race condition",
            "Wrote a failing test for the race",
            "Applied fix: wrap refresh in mutex",
            "Test now passes",
        ]:
            await client.emit(
                act="DO",
                kind="bug.investigation.step",
                body={"description": step},
                thread=thread,
            )

        # Observe + crystallize
        await client.emit(
            act="KNOW",
            kind="bug.investigation.finding",
            body={"topic": "Root cause was lock-free optimistic refresh retry loop."},
            thread=thread,
        )
        await client.emit(
            act="LEARN",
            kind="engineering.insight.concurrency",
            body={"insight": "Token refresh in concurrent contexts must be mutexed"},
            thread=thread,
        )

        # Query back + return summary — one line of context-visible output
        records = await client.query_thread(thread=thread)
        print(f"Bug #342 investigation complete — {len(records)} records in thread {thread}")

asyncio.run(main())

I investigated bug #342 and found the root cause: the auth module's
token refresh logic had a race condition in concurrent refresh attempts.
I applied a mutex around the refresh path, wrote a failing test first,
verified it passes, and recorded 8 records in thread bug-342.

import { Client, Identity } from "@syncropel/sdk";

interface Env {
  SYNCROPEL_URL: string;
  SPL_TOKEN: string;
  AGENT_DID: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const client = new Client({
      endpoint: env.SYNCROPEL_URL,
      identity: Identity.static(env.AGENT_DID),
      apiKey: env.SPL_TOKEN,
    });

    const threadId = "th_webhooks";

    // Record the incoming event
    await client.emit({
      act: "DO",
      kind: "webhook.event.received",
      body: {
        description: "Received webhook",
        event: request.headers.get("x-event"),
        path: new URL(request.url).pathname,
      },
      thread: threadId,
    });

    // Business logic — not Syncropel's concern
    const result = await processWebhook(request);

    // Record the outcome
    await client.emit({
      act: "KNOW",
      kind: "webhook.event.processed",
      body: { topic: "processed", result_code: result.code },
      thread: threadId,
    });

    return new Response(JSON.stringify({ ok: true }), {
      headers: { "content-type": "application/json" },
    });
  },
};

async function processWebhook(request: Request) {
  return { code: "ok" };
}

# Save at end of each work session
result = await client.emit(
    act="LEARN",
    kind="agent.session.checkpoint",
    body={
        "summary": "Completed auth refactor; next: migration tests",
        "learnings": ["mutex pattern worked", "refactor touched 14 files"],
        "next_steps": "Write migration tests",
    },
    thread=session_thread,
)

# Parent agent creates subtasks
for worker in ["agent-a", "agent-b", "agent-c"]:
    child = await client.intend(goal=f"Review module X for {worker}")
    # Each worker reads child.thread_id, does the work,
    # and emits records on that thread. The parent folds
    # them back via query_thread for the join decision.

import httpx

r = httpx.get(
    f"{endpoint}/v1/trust",
    headers={"Authorization": f"Bearer {api_key}"},
)
scores = r.json()["data"]
mine = [s for s in scores if s["actor"] == my_did and s["domain"] == "code"]
if not mine or mine[0]["effective_trust"] < 0.5:
    # Low confidence — emit an INTEND with body.awaits = "actor_decision"
    # so a human reviews before proceeding
    pass

result = await client.emit(...)
if not result.success:
    # Transport error — log locally, retry later, but don't crash
    print(f"Syncropel unreachable: {result.error}")

client = Client(
    endpoint="https://your-host.example.com:9100",
    identity=Identity.static("did:sync:agent:me"),
    api_key=os.environ["SPL_TOKEN"],  # from ~/.syncro/token or env
)