astraform-remote-domain-author-kit

This is the Python author kit for remote-domain.v1.

Brutal truth: an "SDK" alone is not enough. If teams still have to reverse engineer request envelopes, invent their own starter layout, or guess how to prove conformance, you did not ship onboarding. You shipped homework.

This author kit is broader than a thin SDK. It includes:

• reusable protocol constants and envelope builders
• request validation helpers
• an optional FastAPI app factory
• a starter template inside the package

If you only want helper functions, that is the SDK-like layer. The author kit is the whole package around it.

Install

Version 0.2.1 targets published remote-domain contract 1.1.0, including native simulation metadata and customer-scoped read-only tools. Both the release lock and exact PUBLISHED_RELEASE snapshot provenance are packaged under contracts/. See the release notes.

Version 0.2.1 is prepared but not yet published. Use the editable installation below until publication; the pinned package commands apply once 0.2.1 is available.

Protocol helpers only:

pip install astraform-remote-domain-author-kit==0.2.1

FastAPI helper included:

pip install 'astraform-remote-domain-author-kit[fastapi]==0.2.1'

Repository-local development:

pip install -e './remote-domain-author-kit-python[fastapi,test]'

The release workflow installs the exact built wheels into a clean environment, copies the starter with astraform-remote-domain-copy-starter, installs that project with --no-deps, validates its catalog, and runs the domain lifecycle with population catalog plus Policy Wind Tunnel conformance. The same checks run against published PyPI packages before creating the GitHub release. The separate scripts/bootstrap-local-sdk.sh command is available for local development.

Core Usage

from astraform.remote_domain.author_kit.protocol import build_manifest
from astraform.remote_domain.author_kit.protocol import build_projection
from astraform.remote_domain.author_kit.protocol import opaque_state
from astraform.remote_domain.author_kit.protocol import projection_envelope
from astraform.remote_domain.author_kit.protocol import success_envelope
from astraform.remote_domain.author_kit.protocol import validate_request_envelope


def manifest() -> dict:
    return build_manifest(
        domain_id="acme-ops",
        display_name="Acme Operations Domain",
        description="Remote proof domain",
        schema_version="acme-ops.state.v1",
        supported_agent_types=["Operator"],
        supported_interaction_modes=["SIMULATION", "HYBRID"],
        tools=[
            {
                "name": "lookup_case",
                "description": "Look up a case in the remote domain.",
                "inputSchema": {"type": "object", "additionalProperties": False},
            }
        ],
    )


def prepare(request: dict) -> dict:
    validate_request_envelope(
        request,
        expected_domain_id="acme-ops",
        expected_operation="prepare",
        require_idempotency=True,
    )
    state = opaque_state(
        "acme-ops.state.v1",
        {"personaName": "Taylor", "completedWorkCount": 0},
    )
    projection = build_projection(
        runtime_metadata={"domainProfile": "acme"},
        status_view={"completedWorkCount": 0},
        inspection_view={"tasks": []},
    )
    return success_envelope(
        request,
        runtime_identity="acme-ops::Taylor",
        next_state=state,
        projection=projection,
    )


def status(request: dict) -> dict:
    validate_request_envelope(
        request,
        expected_domain_id="acme-ops",
        expected_operation="status",
        require_state=True,
    )
    return projection_envelope(
        request,
        projection=build_projection(
            runtime_metadata={"domainProfile": "acme"},
            status_view={"completedWorkCount": 0},
            inspection_view={"tasks": []},
        ),
    )

Domain providers may add optional evidence_events=[...] to success_envelope(...) or projection_envelope(...). Use that lane for provider-internal evidence that Astraform cannot observe directly, such as a private third-party call or domain-owned policy check.

Native Event Simulation

Use the existing manifest builder; no provider-side manifest patching is needed:

manifest = build_manifest(
    domain_id="acme-ops",
    display_name="Acme Operations",
    description="Provider-owned case reviews",
    schema_version="acme.state.v1",
    supports_domain_system_work_window=True,
    native_simulation={
        "profile": "remote-domain.native-state-transform.v1",
        "providerRevision": "acme-build-42",
        "mutationMode": "PURE_STATE_TRANSFORM",
        "stateScope": "AGENT",
        "workKinds": ["case_review"],
        "maxWindowWorkItems": 1,
        "initialBoundaryMode": "PREPARE_INCLUDES_START",
    },
    tools=[{
        "name": "get_my_cases",
        "description": "Read cases belonging to the configured customer.",
        "revision": "v1",
        "effectClass": "READ_ONLY",
        "scope": "CUSTOMER_CONTEXT",
        "inputSchema": {"type": "object", "additionalProperties": False},
        "outputSchema": {
            "type": "object",
            "additionalProperties": False,
            "properties": {"caseIds": {"type": "array", "items": {"type": "string"}}},
            "required": ["caseIds"],
        },
    }],
)

For partner-owned database state, select profile: remote-domain.native-reference-state.v1, mutationMode: TRANSACTIONAL_REFERENCE_STATE, and initialBoundaryMode: ATTACH_PREPOPULATED_STATE. The remaining declaration fields are shared. The provider remains responsible for its durable data, scoped customer access, idempotency and the declared native execution semantics.

The builder validates native declarations against the embedded canonical schema. A tool carrying revision, effectClass or scope must supply the complete native read-only descriptor, including both schemas and its description. Legacy name/description/inputSchema descriptors remain supported; declaring a legacy tool does not make it eligible for native customer execution. extra_capabilities accepts the same native declaration and receives the same validation. Supply it there or through native_simulation, not both.

Existing envelope helpers carry provider metadata; they do not implement the provider's event processing or advance Astraform's simulation. See the canonical native profile definitions in contracts/remote-domain/v1/profiles/ and the conformance package's validate_native for metadata validation.

FastAPI App Factory

from astraform.remote_domain.author_kit.fastapi import create_fastapi_app

app = create_fastapi_app(
    service=my_remote_domain_service,
    policy_wind_tunnel_service=my_policy_wind_tunnel_service,
)

Every POST route, including Policy Wind Tunnel routes, limits its incoming JSON body to 4 MiB by default. The adapter counts actual streamed bytes before parsing or calling the service, including requests without an accurate Content-Length. Oversized bodies return HTTP 413 with error.code=request_too_large; malformed UTF-8 JSON and non-object bodies return HTTP 400. Empty lifecycle-control bodies remain supported. Set create_fastapi_app(..., max_request_bytes=8 * 1024 * 1024) to use a different positive integer byte limit. This is an admission limit for the complete request, separate from the manifest's maxOpaqueStateBytes setting.

Callback execution and capacity (0.2.1)

The factory dispatches all synchronous domain and Policy Wind Tunnel callbacks through AnyIO worker threads. The HTTP request still waits for the actual result; this does not introduce a background-job or accepted/pending API. Request context variables are copied to the worker, and existing provider errors retain their HTTP status and protocol envelope.

max_concurrent_callbacks is a positive integer, default 1, shared by these SDK callbacks within each app instance/process. This preserves serialized provider calls while allowing health checks and unrelated async routes to run. When every slot is occupied, another SDK callback receives HTTP 503 with error.code=provider_busy, category=transient_failure and retryable=true. It is not queued or invoked. Retry with backoff and the same idempotency key. The health endpoint does not consume a slot; it reports liveness, not spare callback capacity. Requests are still subject to the body limit above.

For a provider whose callbacks and database access support concurrent use:

app = create_fastapi_app(service=my_remote_domain_service, max_concurrent_callbacks=4)

Choose this limit alongside the database connection pool and deployment worker count. Uvicorn's --workers configures server processes in the partner deployment; it is separate from this SDK limit. Each process has its own limit. Custom partner routes and provider-created background jobs are outside the SDK callback limit. Do not raise it without making shared state thread-safe. Even with the default of one, successive calls can use different worker threads; use a connection pool or create/use thread-affine resources inside the callback.

Once admitted, a callback may finish even if the caller cancels or disconnects. Its slot stays occupied until the worker completes. Cancellation cannot undo a database write or forcibly stop synchronous code. Providers must enforce their own database/network timeouts and retain transactional idempotency/replay for uncertain responses. Synchronous service methods remain the supported interface; this change does not add an async provider interface or change simulated-time acknowledgment and advancement rules.

The service object must implement:

• manifest()
• optional population_catalog() for GET /api/remote-domain/v1/population/catalog
• prepare(payload)
• execute_work(payload)
• status(payload)
• inspection(payload)
• shutdown(payload)

population_catalog() must return domain_population_catalog.v1. The FastAPI factory validates the catalog before returning it, so missing archetype fields, bad recipe numbers, unknown archetype references, and malformed eligibility fail before Population Builder treats the provider as usable.

Prefer a file-backed catalog over an inline Python dict. That keeps the domain authoring surface reviewable and conformance-testable:

from importlib import resources

from astraform.remote_domain.author_kit import load_population_catalog


def population_catalog() -> dict:
    catalog_file = resources.files("my_remote_domain").joinpath("population_catalog.json")
    return load_population_catalog(catalog_file)

load_population_catalog(...) reads JSON and runs the same validate_population_catalog(...) checks used by the FastAPI route.

The bundled fastapi-minimal starter also installs a provider-local preflight:

validate-population-catalog

That command validates src/acme_remote_domain/population_catalog.json before the provider starts and prints the exact rejected field when the catalog is malformed. Treat it as the first command after editing archetypes, segment templates, validation rules, or run-path eligibility.

The optional policy_wind_tunnel_service object exposes the remote Wind Tunnel provider routes:

• GET /policy-wind-tunnel/pack
• GET /policy-wind-tunnel/presets
• POST /policy-wind-tunnel/runs
• GET /policy-wind-tunnel/runs/{runId}
• POST /policy-wind-tunnel/runs/{runId}/lifecycle
• GET /policy-wind-tunnel/runs/{runId}/bundle
• GET /policy-wind-tunnel/runs/{runId}/artifacts/{artifactType}
• GET /policy-wind-tunnel/runs/{runId}/artifacts/{artifactType}/readiness
• GET /policy-wind-tunnel/runs/{runId}/evidence-pack
• GET /policy-wind-tunnel/runs/{runId}/evidence-pack/readiness

The readiness routes are metadata-only checks for dashboard download UX. They must prove the same provider-owned artifact/evidence-pack path is readable without materializing the JSON body or ZIP archive.

Policy Expressions

When a domain exposes Policy Wind Tunnel behavior, CEL-compatible PolicyExpression definitions are the portable decision contract. They are for decision gates, eligibility checks, thresholds, and conformance-testable public rules.

They are not the execution engine. Keep state mutation, external calls, proprietary scoring, and side effects inside your service implementation or domain-owned execution target. Use the expression's executionTarget as a stable provider-owned target name.

Python providers should publish the same /policy-wind-tunnel/pack metadata as Java providers. The helper below builds the portable parts of that response:

from astraform.remote_domain.author_kit import build_policy_expression
from astraform.remote_domain.author_kit import build_policy_wind_tunnel_pack_metadata


def wind_tunnel_pack() -> dict:
    return build_policy_wind_tunnel_pack_metadata(
        capabilities={"remoteProviderReady": True, "evidenceExport": True},
        control_schema={
            "type": "object",
            "properties": {
                "blockerCount": {"type": "integer", "minimum": 0},
            },
        },
        ui_schema={
            "fields": [
                {"name": "blockerCount", "widget": "number", "label": "Blockers"},
            ],
        },
        outcome_schema={
            "schemaVersion": "domain_outcome_schema.v1",
            "metricDefinitions": [
                {
                    "metricId": "blockerCount",
                    "label": "Blockers",
                    "unit": "count",
                    "evidence": "domain-state",
                },
            ],
        },
        domain_labels={
            "actorSingular": "campaign",
            "actorPlural": "campaigns",
            "simulatedActorsLabel": "Simulated campaigns",
        },
        policy_expressions=[
            build_policy_expression(
                expression_id="acme.blockers.acceptable",
                expression="metrics.blockerCount <= 10",
                execution_target="acme-policy-service",
                description="Blockers must stay under launch threshold.",
            ),
        ],
    )

Keep policy expression contract shape in the public contract bundle and SDK tests so Java and Python providers evaluate the same portable subset.

Starter Template

Copy the bundled FastAPI starter from the installed author kit:

astraform-remote-domain-copy-starter acme-remote-domain
cd acme-remote-domain
pip install --no-deps -e .
validate-population-catalog
remote-domain-conformance --app acme_remote_domain.main:app --domain-id acme-remote --population-catalog

The underlying template resource lives under:

• astraform/remote_domain/author_kit/templates/fastapi-minimal

It is intentionally boring. That is a feature. Teams need a truthful starting point, not a framework demo that hides the protocol.

Publishing Status

Release validation starts with the standalone package tests:

pytest remote-domain-author-kit-python/tests remote-domain-conformance-python/tests

PyPI artifacts are immutable. Do not rerun a publish for an already published version; run smoke-only verification or bump the SDK version.

Public package note: PyPI distributions expose this SDK implementation. Keep host runtime internals and domain-private logic out of this package.