Python

sdks / v1

Shared WASM, separate Python host

agindow 0.1.0a0, unpublished. Python 3.11+, pinned Wasmtime, and identical WASM bytes to Node. Build the shared guest first and install the local wheel. No separate Python detector or framework adapter is shipped.

Decorators and ContextVar scopes

Agindow.guard works as a decorator; wrap accepts callables. with sdk.context and async with sdk.context apply ContextVar overrides only to that SDK’s guarded calls, remain isolated across tasks and freeze for call completion. Sync stays sync, async stays async, and cancellation propagates. The example validates both scopes and original object identity.

Readiness and configuration

sdk.ready is a synchronous boolean after loading, unlike Node’s Promise; sdk.load_error retains failure. configure changes future calls. Snake_case keywords resolve explicit > environment > nearest JSON .agindowrc > defaults. NullEngine and observe are defaults. sdk.close releases the guest; later calls bypass.

Remote and framework boundaries

guard_remote and guard_remote_async are explicit HTTP APIs, separate from local decorators. There are no Python framework adapters, built-in report CLI, distributed replay or hosted delivery. The shared default guest has no selective repair output.

Run the fixture

Save the Python example as example.py in cells/cell-relay/python, then run uv run --frozen python example.py. Sync, async and closed-engine outcomes are synthetic local evidence.

Continue

Local interfaces are usable before registry publication. Choose the transport explicitly.

import asyncio
from agindow import Agindow
sdk = Agindow(mode="observe")
payload = {"id": "synthetic-customer"}
@sdk.guard(tool="vendor.read", operation="read")
def read() -> dict[str, str]:
    return payload
@sdk.guard(tool="vendor.read", operation="read")
async def read_async() -> dict[str, str]:
    return payload
with sdk.context(mode="observe"):
    assert read() is payload
async def main() -> None:
    async with sdk.context(mode="observe"):
        assert await read_async() is payload
asyncio.run(main())
# A closed engine is a labelled availability fixture, not vendor drift.
sdk.close()
assert read() is payload
assert len([event for event in sdk.recent() if event.type == "bypass"]) == 1
print("Synthetic fixture: sync/async identity retained; closed engine bypassed once.")