Skip to content
AIWS / SDK

Python integration

Python 3.11 or newer is required. Install the downloaded aiws_sdk-0.3.0-py3-none-any.whl, or run python -m pip install ./packages/python from the repository root. jsonschema is the runtime dependency. This is a native Python implementation and is not published to PyPI in this release.

Runnable coordinator

Section titled “Runnable coordinator”

Execute an adapter through the trusted coordinator · Executable local adapter demonstration; fixed demo identity and fixture evidence

import { readFileSync, writeFileSync, mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { parseContract, parseCommand, canonical, assessSuccess } from '@aiws/sdk';
import { SqliteStore } from '@aiws/sdk/sqlite';
import { Coordinator, type EffectAdapter } from '@aiws/sdk/runtime';
// A local demonstration. Replace this fixed demo identity with authenticated policy.
const demo = JSON.parse(readFileSync('examples/review.json', 'utf8'));
const directory = mkdtempSync(join(tmpdir(), 'aiws-review-'));
const store = new SqliteStore(join(directory, 'mission.db'), parseContract(canonical(demo.contract)));
const coordinator = new Coordinator(store, async () => ({ allowed: true, policy: 'ALLOW', mandatoryChecksOk: true,context:{actor:"demo:operator",policyRevision:"policy:1",decisionClass:"DETERMINISTIC"} }), () => '10');
const adapter: EffectAdapter = {
    async execute(operation) { writeFileSync(join(directory, 'review.txt'), canonical(operation.action.payload), { flag: 'wx' }); return { effect: 'CONFIRMED_APPLIED', actualCost: '3' }; },
    async reconcile(operation) { try {
        return { effect: readFileSync(join(directory, 'review.txt'), 'utf8') === canonical(operation.action.payload) ? 'CONFIRMED_APPLIED' : 'UNKNOWN', actualCost: '3' };
    }
    catch {
        return { effect: 'UNKNOWN', actualCost: '0' };
    } }
};
try {
    for (const raw of demo.commands) {
        const command = parseCommand(canonical(raw));
        if (command.type === 'dispatch')
            await coordinator.dispatch(command.attemptId, adapter, command.nodeId);
        else
            await coordinator.apply(command);
    }
    console.log(JSON.stringify({ directory, ...assessSuccess(store.snapshot(), 'r') }));
    writeFileSync(join(directory, 'audit.json'), store.exportAudit());
}
finally {
    store.close();
}

The full program uses a local adapter, fixed trusted demo identity and fixture evidence. Replace those three boundaries with real identity, a scoped adapter and measured validation evidence before using this pattern in production. The tab set includes equivalent Rust, TypeScript and Python implementations.

Protocol-bound workflow correlation

Section titled “Protocol-bound workflow correlation”

The current repository source adds a small protocol-correlation helper after the previously built SDK 0.3.0 binary baseline. It records the exact Praxis binding revision and remote identity alongside the AIWS work/run/operation/attempt identity. It does not implement MCP, A2A or AG-UI transport behavior and always validates authoritative: false.

Correlate governed work with a protocol binding · Executable source-only SDK correlation example; no protocol client or authority is created

import assert from 'node:assert/strict';
import {createMission,reduce,protocolCorrelation,type Contract} from '@aiws/sdk';


const contract:Contract={id:'contract:protocol',missionId:'mission:protocol',aiwsEdition:'0.4',profile:'finite-v1',budget:'20',deadline:'1000',criteria:['remote-result-reviewed'],actions:['remote.invoke'],resources:['protocol:mcp'],requiresApproval:false,maxAttempts:'2',maxAuthorizationAgeMs:'100',features:[]};
let state=createMission(contract);
state=reduce(state,{type:'startRun',runId:'run:1'},'10');
state=reduce(state,{type:'grant',grant:{id:'grant:1',subject:'agent:1',profile:'finite-v1',actions:['remote.invoke'],resources:['protocol:mcp'],notBefore:'0',expiresAt:'1000',limit:'20',canDelegate:false,depth:'0'}},'10');
state=reduce(state,{type:'admit',runId:'run:1',operationId:'operation:1',attemptId:'attempt:1',grantId:'grant:1',subject:'agent:1',action:{capability:'remote.invoke',resource:'protocol:mcp',payload:{tool:'echo'},preconditions:{}},amount:'5',authorizationCheckedAt:'10',policy:'ALLOW',mandatoryChecksOk:true},'10');


const correlation=protocolCorrelation(
  {bindingId:'mcp:official-v2',bindingRevision:'1',protocol:'MCP',protocolVersion:'2026-07-28',manifestDigest:'a'.repeat(64),capabilityDigest:'b'.repeat(64)},
  {workOrderId:'mission:protocol',runId:'run:1',taskId:null,operationId:'operation:1',attemptId:'attempt:1',remoteIdentityRef:'mcp:praxis-m9-fixture@1.0.0',remoteOperationId:null}
);
assert.equal(correlation.operation.operationId,'operation:1');
assert.equal(correlation.authoritative,false);
assert.equal(state.operations['operation:1'].runId,'run:1');
console.log('PASS: protocol-correlation');

Use official protocol SDKs and the Praxis adapters for wire behavior. Use this helper only for portable correlation in application code, evidence indexes or integration metadata.

Integration details

Section titled “Integration details”

Core records are validated dictionaries, not Pydantic models or mutable workflow classes. Import parse_contract, parse_command, reduce and AiwsError from aiws or aiws.core, and use aiws.runtime, aiws.sqlite and aiws.observability for the runtime boundaries.

The authorizer is a callable returning allowed, policy, mandatoryChecksOk and context. The clock returns str(time.time_ns() // 1_000_000) in a real host. Demo clocks are deterministic test inputs and must not become production authorization clocks. Adapter execute and reconcile return dictionaries containing effect and actualCost.

The SDK is synchronous. If an async web framework is used, execute blocking work in a suitable worker with its own connection lifecycle. Do not move an existing SQLite connection freely between threads; create and use it under the host’s defined ownership rules. A with SqliteStore(…) block closes the connection, but that does not imply cancellation of an external job.

Python integers are unbounded, but the cross-language wire format is deliberately bounded. Do not bypass the safe-number rule just because Python can decode a large integer. Costs and timestamps remain canonical decimal strings. Catch AiwsError and inspect code at an integration boundary; never turn a generic Exception into automatic permission to retry an external action.

Diagnostics and next steps

Section titled “Diagnostics and next steps”

Use the API mapping to translate native calls without changing camelCase wire keys. Use observability for exporter configuration and recovery for interrupted effects. All examples are included as individual executable source files under examples/guide/.

The supplied CLI validates contracts and graphs and replays audit records. Its observe projection uses a fixed test timestamp and is not a live monitoring service. Run operationalSummary/operational_summary with a trusted current timestamp in your application instead.