# Tutorial: run a managed agent

> Register an agent, publish its definition and run it once as a free Test simulation, with the TypeScript or Python client.

Section: Synapse and Locus. Web page: https://docs.katara.com/products/tutorial-managed-agent/

This tutorial builds an agent that Katara runs, then starts one run in the Test environment as a **simulation**: a fixed scenario that makes no model call, spends no credits and has no effect outside Katara. Everything else is what paid work uses: the agent's identity, its approved definition, a budget, grants and an admitted run you can follow to the end.

The calls and request shapes below are checked against the API contract and both clients. [Product availability](https://docs.katara.com/products/availability.txt) has the checks that tell you whether runs execute where you call them.

**Get Started** in the portal sets up an agent in one step, with a first run that is paid Test Compute; see [Set up a first agent](https://docs.katara.com/start/first-agent.txt). This page is the same path taken one call at a time, without spending.

## Who does what

Two steps need a person signed in to the portal: approving content and funding a budget. An API key cannot do either. Your script does the rest with that same person's API key. One person does all of it, because a grant is issued by whoever funded the budget it draws on.

| Step | Who | How |
| --- | --- | --- |
| Create a Test API key | Person | **API Keys** in the portal |
| Make sure the environment has a content store | Person | **Synapse → Storage** |
| Register the agent and upload its definition and input | Script | `createServiceAccount`, `createAgentIdentity`, content uploads |
| Approve the four uploads | Person | **Synapse → Storage**, then **Review and Approve** on each |
| Publish the agent version | Script | `listSimulationFixtures`, `publishAgentVersion` |
| Fund a simulation budget | Person | **Locus → Budgets & Grants → Fund a Budget**, usage **Simulation** |
| Grant the agent its authority, start the run and follow it | Script | `issueGrant`, `enableContentAccess`, `createRun`, `getRun` |

The script stops at each step a person has to take and says what to do. Run it again afterwards and it continues.

## Before you start

- You are an owner or admin of an organization and use its **Test** environment.
- The environment has a content store. Under **Synapse → Storage**, create Katara storage or connect a bucket. `listContentStores` / `list_content_stores` shows what exists.
- A personal Test API key from **API Keys** (not a service-account key) with these scopes:
  - `org:read` to read the environment
  - `org:service_accounts:write` to create the agent's service account
  - `locus:grants:read` and `locus:grants:write` to register the agent and issue its grants
  - `synapse:memory:read` and `synapse:memory:write` to upload content and share it with the agent
  - `synapse:agents:read` and `synapse:agents:write` to read fixtures and publish the version
  - `synapse:runs:create` and `synapse:runs:read` to start and read the run

  A key cannot grant more than it holds, so it carries every permission the grants name.
- The TypeScript client `@katara/platform` with Node.js, or the Python client `katara-platform`. Neither is on npm or PyPI yet; use the copy supplied for your environment.
- `KATARA_API_KEY`, `KATARA_ORG_ID` and `KATARA_ENVIRONMENT_ID` set in the environment, the last one to the Test environment's ID.

## The script

Save this as `agent.mts` and run it with a TypeScript runner, for example `npx tsx agent.mts`.

The script writes `agent-tutorial.json` beside itself. It holds every ID the script creates and every request key it uses, saved before the request is sent. If a reply is lost, running the script again sends the same request with the same key and gets the original result, never a second agent, upload or run.

```ts
import { createHash, randomUUID } from 'node:crypto';
import { existsSync, readFileSync, writeFileSync } from 'node:fs';
import { ProductClient, PlatformError, uploadContent, type ProductContentRef, type ProductScope } from '@katara/platform';

const client = new ProductClient({ baseUrl: 'https://api.katara.com', credential: () => process.env.KATARA_API_KEY ?? null });
const scope: ProductScope = { org_id: process.env.KATARA_ORG_ID!, environment_id: process.env.KATARA_ENVIRONMENT_ID! };

// Each value is saved before the request that uses it, so a repeated request is identical.
const file = 'agent-tutorial.json';
const state: Record<string, any> = existsSync(file) ? JSON.parse(readFileSync(file, 'utf8')) : {};
const save = () => writeFileSync(file, JSON.stringify(state, null, 2));
function remember<T>(name: string, make: () => T): T {
  if (!(name in state)) { state[name] = make(); save(); }
  return state[name];
}
async function once<T>(name: string, send: (requestKey: string) => Promise<T>): Promise<T> {
  const key = remember(`${name}.key`, () => `tutorial-${randomUUID()}`);
  if (state[`${name}.result`] === undefined) { state[`${name}.result`] = await send(key); save(); }
  return state[`${name}.result`];
}
function stop(message: string): never { console.log(message); process.exit(0); }

async function upload(name: string, mediaType: string, text: string, storeId: string): Promise<ProductContentRef> {
  const bytes = new TextEncoder().encode(text);
  const started = await once(`${name}.upload`, key => client.beginContentUpload(scope, {
    store_id: storeId, object_id: remember(`${name}.object`, randomUUID), previous_version: 0,
    integrity: 'sha256:' + createHash('sha256').update(bytes).digest('hex'), media_type: mediaType, size_bytes: bytes.length, name,
  }, key));
  await once(`${name}.complete`, async key => {
    // After a lost reply the bytes may already be there; send them only while the upload is open.
    const history = await client.contentHistory(scope, started.content);
    if (history.data.find(v => v.content.version === started.content.version)?.state === 'uploading') {
      await uploadContent(await client.issueContentUpload(scope, started.content), started.content, bytes);
    }
    return client.completeContentUpload(scope, started.content, key);
  });
  return started.content;
}

async function main() {
  const environment = await client.getEnvironment(scope);
  if (environment.mode !== 'test') stop('Set KATARA_ENVIRONMENT_ID to a Test environment.');
  const stores = (await client.listContentStores(scope)).data.filter(s => s.state === 'active');
  if (!('target' in state) && !stores.length) stop('The environment has no content store. Create one under Synapse → Storage.');
  const target = remember('target', () => ({ store_id: stores[0].id, store_version: stores[0].version, key_id: stores[0].key_id, key_version: stores[0].key_version, retention_seconds: 86400 }));

  // The agent: a service account, and an identity for it in this environment.
  const account = await once('service-account', key => client.createServiceAccount(scope.org_id, 'Tutorial agent', key));
  const pins = remember('pins', () => ({ epoch: environment.authority_epoch, generation: environment.reset_generation }));
  const identity = await once('identity', key => client.createAgentIdentity(scope, {
    expected_environment_epoch: pins.epoch, expected_reset_generation: pins.generation,
    name: 'Tutorial agent', service_account_id: account.resource_id, execution_kind: 'managed',
  }, key));
  const agentId = identity.resource_id;

  // Its definition and the run's input, as content.
  const schema = (field: string) => JSON.stringify({ $schema: 'https://json-schema.org/draft/2020-12/schema', type: 'object', properties: { [field]: { type: 'string' } }, required: [field], additionalProperties: false });
  const documents = {
    instructions: await upload('instructions.md', 'text/markdown', 'Answer the task in one short paragraph.\n', target.store_id),
    input_schema: await upload('input-schema.json', 'application/json', schema('task'), target.store_id),
    output_schema: await upload('output-schema.json', 'application/json', schema('answer'), target.store_id),
    input: await upload('input.json', 'application/json', JSON.stringify({ task: 'Say hello.' }), target.store_id),
  };
  const waiting: string[] = [];
  for (const [name, ref] of Object.entries(documents)) {
    const history = await client.contentHistory(scope, ref);
    if (history.data.find(v => v.content.version === ref.version)?.state !== 'approved') waiting.push(name);
  }
  if (waiting.length) stop(`Approve ${waiting.join(', ')} under Synapse → Storage, then run this again.`);

  // The version, pinned to the simulation scenario that succeeds.
  const fixture = (await client.listSimulationFixtures(scope)).data.find(f => f.id === 'katara.test.success.v1');
  if (!fixture) stop('The success scenario is not in this environment\'s simulation catalog.');
  const published = await once('version', key => client.publishAgentVersion(scope, {
    agent_id: agentId, expected_version: 0,
    definition: {
      schema_version: 1, name: 'Tutorial agent', instructions_ref: documents.instructions, model_policy: fixture.model_policy,
      tool_versions: [], knowledge_collections: [], approval_policy_versions: [],
      input_schema_ref: documents.input_schema, output_schema_ref: documents.output_schema,
      limits: { max_steps: 20, max_decisions: 30, max_plan_changes: 10, max_children: 0, max_depth: 0, max_active_seconds: 300, max_elapsed_seconds: 3600, max_approval_wait_seconds: 600, max_output_bytes: 1048576 },
    },
  }, key));
  const signed = await client.getAgentVersion(scope, published.resource_id);
  const version = { id: signed.bundle.id, version: signed.bundle.revision, integrity: signed.digest };

  // Authority: a grant to run, and read access to the definition. Both draw on the budget.
  const budgetId = process.env.KATARA_BUDGET_ID;
  if (!budgetId) stop('Fund a simulation budget under Locus → Budgets & Grants, set KATARA_BUDGET_ID to its ID and run this again.');
  const agent = (await client.listAgentIdentities(scope)).data.find(i => i.id === agentId)!;
  const authority = remember('authority', () => ({ epoch: agent.authority_epoch, expires_at: new Date(Date.now() + 7 * 86400_000).toISOString() }));
  const runGrant = await once('run-grant', key => client.issueGrant(scope, {
    expected_agent_epoch: authority.epoch, agent_id: agentId, budget_id: budgetId, audience: 'katara:synapse',
    actions: ['synapse:runs:create', 'synapse:runs:read', 'synapse:agents:read'], resources: [agentId],
    expires_at: authority.expires_at, remaining_depth: 0,
  }, key));
  const readGrant = await once('content-access', key => client.enableContentAccess(scope, agentId, {
    agent_version: version, budget_id: budgetId, expires_at: authority.expires_at, expected_agent_epoch: authority.epoch,
  }, key));

  // The run.
  const admitted = await once('run', key => client.createRun(scope, {
    agent_version: version, grant_id: runGrant.resource_id,
    workload_grants: [{ id: runGrant.resource_id, audience: 'katara:synapse' }, { id: readGrant.resource_id, audience: 'katara:synapse' }],
    input_ref: documents.input, output_target: target, budget_id: budgetId, usage_class: 'simulation',
  }, key));
  console.log('run', admitted.resource_id);
  for (let i = 0; i < 60; i++) {
    const run = await client.getRun(scope, admitted.resource_id);
    const snapshot = run.observation?.snapshot;
    console.log(run.status, run.reason ?? '', snapshot ? `steps ${snapshot.steps}, decisions ${snapshot.decisions}` : '');
    if (['completed', 'failed', 'cancelled', 'reconciliation_required'].includes(run.status)) return;
    await new Promise(resolve => setTimeout(resolve, 5000));
  }
  console.log('The run has not finished. Run this again to keep following it.');
}

main().catch(error => {
  if (error instanceof PlatformError) console.error(error.status, error.code, error.message, 'request', error.requestId);
  else console.error(error);
  process.exit(1);
});
```

## What you see

The first time, the script registers the agent, uploads four documents and stops:

```text
Approve instructions, input_schema, output_schema, input under Synapse → Storage, then run this again.
```

Open each one under **Synapse → Storage** and select **Review and Approve**. The person whose key uploaded them is the one who can approve them. Then fund a budget under **Locus → Budgets & Grants → Fund a Budget**: choose **Simulation**, any limit, and caps of that limit for inference and service. A simulation budget holds no credits. Copy its ID into `KATARA_BUDGET_ID` and run the script again.

It publishes version 1, issues the two grants and starts the run. `createRun` answers `202` with an operation whose `resource_id` is the run ID. Admission means the run is recorded as `queued`; it does not mean anything has picked it up. When execution is available, the run moves on and ends like this, though which intermediate statuses you see depends on timing:

```text
run 4f0e5a2c-…
queued
running
completed completed steps 1, decisions 2
```

The success scenario makes one simulated model call and one decision to finish, so `steps` is 1 and `decisions` is 2. A simulation produces no output: `observation.snapshot.output` is absent, and the [completed-output recipe](https://docs.katara.com/products/sdks.txt#read-the-completed-runs-output) returns `null` for it.

## When a run does not finish

| What you see | What it means | What to do |
| --- | --- | --- |
| `queued`, for minutes | Nothing has picked the run up | Check [managed execution](https://docs.katara.com/products/availability.txt#managed-execution). Keep the run ID; do not start another |
| `failed`, reason `authority_denied` | The run lacked something it needed: a grant, or read access to its definition | Check both grants are in `workload_grants` and have not expired or been revoked |
| `failed`, reason `queue_expired` | The run waited in the queue too long | Start a new run when execution is available |
| `reconciliation_required`, reason `execution_unknown` | A step may or may not have taken effect | Do not repeat it. See [recovery](#recover-from-an-uncertain-response) |
| `createRun` refused, `403` `product_access_unavailable` | The environment has no Synapse access for simulation yet | Wait and try again; access for a new organization can take a moment |
| `createRun` refused, `409` `conflict` | Something the request depends on changed after it was read. The message names the check, for example `the output store or its key changed` | Read the current state and start again with new keys |

`runEvents` / `run_events` returns every status change with a sequence number, and [Watching a run](https://docs.katara.com/products/run-events.txt) explains resuming from the last one you saw. [Define, test and inspect an agent](https://docs.katara.com/products/runs.txt#inspect-a-run-trace) covers the trace, artifacts and charges.

The other two scenarios show the failure paths without waiting for a real failure. Publish a version with the `model_policy` of `katara.test.denied.v1` and its run ends `failed` with reason `authority_denied`. With `katara.test.interrupted.v1` it ends `reconciliation_required` with reason `execution_unknown`, which is what an uncertain step looks like. An interrupted run keeps holding one of the environment's active run places while it waits for reconciliation.

## Recover from an uncertain response

If a request returns no answer, or an error you cannot place, send it again unchanged: the same body under the same request key. The server answers with the original receipt and does nothing twice. The script does this for you, because it saved each key and body before sending them. A request sent under a new key is a new request: a second agent, a second upload, a second run.

`409` `conflict` means the key was used for a different body, or something the request depends on changed after you read it. Do not retry it. Read the current state, review the change and send it under a new key.

`reconciliation_required` is not an error to retry. Keep the run ID. An owner or admin records what happened through the [reconciliation operation](https://docs.katara.com/products/closing-an-organization.txt#resolve-a-run-awaiting-reconciliation); recording it charges, refunds and retries nothing.

## The same script in Python

Save this as `agent.py` and run `python agent.py`. It keeps the same state file, so do not switch languages half way through.

```python
import hashlib, json, os, sys, time, uuid
from datetime import datetime, timedelta, timezone
from katara_platform import PlatformError, ProductClient, upload_content

client = ProductClient('https://api.katara.com', lambda: os.environ.get('KATARA_API_KEY'))
org, env = os.environ['KATARA_ORG_ID'], os.environ['KATARA_ENVIRONMENT_ID']

FILE = 'agent-tutorial.json'
state = json.load(open(FILE)) if os.path.exists(FILE) else {}

def save():
    with open(FILE, 'w') as f:
        json.dump(state, f, indent=2)

def remember(name, make):
    if name not in state:
        state[name] = make()
        save()
    return state[name]

def once(name, send):
    key = remember(name + '.key', lambda: 'tutorial-' + str(uuid.uuid4()))
    if state.get(name + '.result') is None:
        state[name + '.result'] = send(key)
        save()
    return state[name + '.result']

def stop(message):
    print(message)
    sys.exit(0)

def upload(name, media_type, text, store_id):
    data = text.encode()
    started = once(name + '.upload', lambda key: client.begin_content_upload(org, env, {
        'store_id': store_id, 'object_id': remember(name + '.object', lambda: str(uuid.uuid4())), 'previous_version': 0,
        'integrity': 'sha256:' + hashlib.sha256(data).hexdigest(), 'media_type': media_type, 'size_bytes': len(data), 'name': name,
    }, key))
    def complete(key):
        # After a lost reply the bytes may already be there; send them only while the upload is open.
        history = client.content_history(org, env, started['content'])['data']
        if next((v['state'] for v in history if v['content']['version'] == started['content']['version']), None) == 'uploading':
            upload_content(client.issue_content_upload(org, env, started['content']), started['content'], data)
        return client.complete_content_upload(org, env, started['content'], key)
    once(name + '.complete', complete)
    return started['content']

def main():
    environment = client.get_environment(org, env)
    if environment['mode'] != 'test':
        stop('Set KATARA_ENVIRONMENT_ID to a Test environment.')
    stores = [s for s in client.list_content_stores(org, env)['data'] if s['state'] == 'active']
    if 'target' not in state and not stores:
        stop('The environment has no content store. Create one under Synapse → Storage.')
    target = remember('target', lambda: {'store_id': stores[0]['id'], 'store_version': stores[0]['version'],
                                         'key_id': stores[0]['key_id'], 'key_version': stores[0]['key_version'], 'retention_seconds': 86400})

    account = once('service-account', lambda key: client.create_service_account(org, 'Tutorial agent', key))
    pins = remember('pins', lambda: {'epoch': environment['authority_epoch'], 'generation': environment['reset_generation']})
    identity = once('identity', lambda key: client.create_agent_identity(org, env, {
        'expected_environment_epoch': pins['epoch'], 'expected_reset_generation': pins['generation'],
        'name': 'Tutorial agent', 'service_account_id': account['resource_id'], 'execution_kind': 'managed',
    }, key))
    agent_id = identity['resource_id']

    def schema(field):
        return json.dumps({'$schema': 'https://json-schema.org/draft/2020-12/schema', 'type': 'object',
                           'properties': {field: {'type': 'string'}}, 'required': [field], 'additionalProperties': False})
    documents = {
        'instructions': upload('instructions.md', 'text/markdown', 'Answer the task in one short paragraph.\n', target['store_id']),
        'input_schema': upload('input-schema.json', 'application/json', schema('task'), target['store_id']),
        'output_schema': upload('output-schema.json', 'application/json', schema('answer'), target['store_id']),
        'input': upload('input.json', 'application/json', json.dumps({'task': 'Say hello.'}), target['store_id']),
    }
    waiting = [name for name, ref in documents.items()
               if next((v['state'] for v in client.content_history(org, env, ref)['data'] if v['content']['version'] == ref['version']), None) != 'approved']
    if waiting:
        stop('Approve ' + ', '.join(waiting) + ' under Synapse → Storage, then run this again.')

    fixture = next((f for f in client.list_simulation_fixtures(org, env)['data'] if f['id'] == 'katara.test.success.v1'), None)
    if fixture is None:
        stop("The success scenario is not in this environment's simulation catalog.")
    published = once('version', lambda key: client.publish_agent_version(org, env, {
        'agent_id': agent_id, 'expected_version': 0,
        'definition': {
            'schema_version': 1, 'name': 'Tutorial agent', 'instructions_ref': documents['instructions'], 'model_policy': fixture['model_policy'],
            'tool_versions': [], 'knowledge_collections': [], 'approval_policy_versions': [],
            'input_schema_ref': documents['input_schema'], 'output_schema_ref': documents['output_schema'],
            'limits': {'max_steps': 20, 'max_decisions': 30, 'max_plan_changes': 10, 'max_children': 0, 'max_depth': 0, 'max_active_seconds': 300,
                       'max_elapsed_seconds': 3600, 'max_approval_wait_seconds': 600, 'max_output_bytes': 1048576},
        },
    }, key))
    signed = client.get_agent_version(org, env, published['resource_id'])
    version = {'id': signed['bundle']['id'], 'version': signed['bundle']['revision'], 'integrity': signed['digest']}

    budget_id = os.environ.get('KATARA_BUDGET_ID')
    if not budget_id:
        stop('Fund a simulation budget under Locus → Budgets & Grants, set KATARA_BUDGET_ID to its ID and run this again.')
    agent = next(i for i in client.list_agent_identities(org, env)['data'] if i['id'] == agent_id)
    authority = remember('authority', lambda: {'epoch': agent['authority_epoch'],
                                               'expires_at': (datetime.now(timezone.utc) + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%SZ')})
    run_grant = once('run-grant', lambda key: client.issue_grant(org, env, {
        'expected_agent_epoch': authority['epoch'], 'agent_id': agent_id, 'budget_id': budget_id, 'audience': 'katara:synapse',
        'actions': ['synapse:runs:create', 'synapse:runs:read', 'synapse:agents:read'], 'resources': [agent_id],
        'expires_at': authority['expires_at'], 'remaining_depth': 0,
    }, key))
    read_grant = once('content-access', lambda key: client.enable_content_access(org, env, agent_id, {
        'agent_version': version, 'budget_id': budget_id, 'expires_at': authority['expires_at'], 'expected_agent_epoch': authority['epoch'],
    }, key))

    admitted = once('run', lambda key: client.create_run(org, env, {
        'agent_version': version, 'grant_id': run_grant['resource_id'],
        'workload_grants': [{'id': run_grant['resource_id'], 'audience': 'katara:synapse'}, {'id': read_grant['resource_id'], 'audience': 'katara:synapse'}],
        'input_ref': documents['input'], 'output_target': target, 'budget_id': budget_id, 'usage_class': 'simulation',
    }, key))
    print('run', admitted['resource_id'])
    for _ in range(60):
        run = client.get_run(org, env, admitted['resource_id'])
        snapshot = (run.get('observation') or {}).get('snapshot') or {}
        print(run['status'], run.get('reason', ''), f"steps {snapshot['steps']}, decisions {snapshot['decisions']}" if snapshot else '')
        if run['status'] in ('completed', 'failed', 'cancelled', 'reconciliation_required'):
            return
        time.sleep(5)
    print('The run has not finished. Run this again to keep following it.')

try:
    main()
except PlatformError as error:
    print(error.status, error.code, error.message, 'request', error.request_id, file=sys.stderr)
    sys.exit(1)
```

## Simulation, Test Compute and Live

The script runs the cheapest of the three usage classes. The other two change these parts of it:

| | Simulation | Test Compute | Live |
| --- | --- | --- | --- |
| Environment | Test | Test | Live |
| Model | A scenario from `listSimulationFixtures`; no model is called | A real model, chosen with `listAgentModels` and pinned with `resolveModelPolicy` | The same |
| Budget | Simulation; no credits are held | Credits are held from the payer's balance when it is funded | The same |
| `createRun` | `usage_class: "simulation"` | `usage_class: "test_compute"` and `paid_test_consent_id`, the consent that funds the budget | `usage_class: "live"` |
| Grants | Run, and read access to the definition | Also a `katara:locus` grant with `cortex:inference` for model calls | The same |
| Tool, code and browser effects | None | Refused by policy | Allowed under the active policy, approvals and the budget |
| Output | None | The agent's answer, an artifact with `untrusted` provenance | The same |

Funding Test Compute or Live needs `live_funding` to be `true` on `listEnvironments`; when it is `false` the deployment takes no credits. A paid run's charges are on `runCharges` / `run_charges` while it runs and on the [organization statement](https://docs.katara.com/products/billing-statements.txt) afterwards. [Define, test and inspect an agent](https://docs.katara.com/products/runs.txt) covers model policies, inputs and the run's authority in full.