Tutorial: run a 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 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. 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.

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:

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:

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 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. 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
`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 explains resuming from the last one you saw. Define, test and inspect an agent 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; 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.

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 afterwards. Define, test and inspect an agent covers model policies, inputs and the run’s authority in full.