astraform-remote-domain-author-kit

This Python partner SDK contains provider helpers for remote-domain.v1 services and the generated platform client for configuring customers, launching simulations and retrieving results. Both are supplied by the existing astraform-remote-domain-author-kit package. Conformance remains separately installed provider validation tooling.

The package includes:

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

Install

Version 0.3.1 is prepared but not yet published. It fixes participation response decoding and preserves shared fields in normalized A2A receipts. The existing package still combines domain provider helpers and the platform API client. Both contract inputs remain on the published shared 1.2.0 release. Release locks and actual input provenance are packaged with the runtime resources; explicit local contract builds record LOCAL_PRERELEASE. See the 0.3.1 release notes (repository access required).

0.3.0 remains available on PyPI and has the response-decoding defects fixed in 0.3.1. After 0.3.1 is published, install with Python 3.12 or newer; Java and Maven are unnecessary for a published installation:

python -m pip install 'astraform-remote-domain-author-kit[fastapi]==0.3.1'

Omit [fastapi] if you only call Astraform. Conformance remains a separately installed validation tool; no extra platform-client distribution is required.

Repository-local development requires Python 3.12+, Java 21 and Maven:

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.

Platform API Client

The platform client is generated inside this package, not a second distribution. The author-kit wheel and source archive contain astraform_platform_client, its runtime dependencies and snapshot provenance. Both install without Java/Maven.

from astraform_platform_client import ApiClient, Configuration
from astraform_platform_client.api.agent_catalog_api import AgentCatalogApi
from astraform_platform_client.api.agent_journeys_api import AgentJourneysApi
from astraform_platform_client.api.simulation_runs_api import SimulationRunsApi

client = ApiClient(Configuration(host="http://localhost:8484", retries=0))
catalog = AgentCatalogApi(client)
journeys = AgentJourneysApi(client)
runs = SimulationRunsApi(client)

Use the Astraform API origin, without /api or /api/dashboard; generated methods include canonical /api/... paths. The examples use http://localhost:8484, the current Gemini/Ollama development profile address. Set the actual gateway origin for your deployment. Replaying a launch requires the same request ID and unchanged request body. Do not add defaults to saved requests: omission and explicit zero can have different identities.

The generated surface follows the Experiment OpenAPI specification. Dashboard exposes Agent Catalog, Agent Journeys, Simulation Events/capability discovery and Simulation Runs read, ledger, readiness and export operations. Other generated Experiment/internal write operations require their own service endpoint; they are not all exposed by Dashboard. An updated Dashboard must provide the canonical aliases. Packaging does not change platform permissions or route exposure.

Create, publish, discover and launch

Start with the existing marketing sample-configuration.json (requires repository access). Save it in your working directory. It supplies catalogDraftRequest and launchRequestTemplate, including the human persona, model binding, facts, domain scheduling and goal-evidence policy. Configure the local-customer model, marketing-ops domain and creative-agency partner in the deployment before running this example. The customer-simulation guide requires platform repository access and explains those bindings, persona/goals, customer tools, partner-owned state and evidence. Package installation alone does not configure them; obtain guide/sample access from your deployment owner.

This code uses the sample's exact request structure. It discovers the configured domain, creates a customer definition, publishes its current draft and launches that immutable revision. Use it once for a new run:

import json
from pathlib import Path
from uuid import uuid4

from astraform_platform_client import ApiClient, Configuration
from astraform_platform_client.api.agent_catalog_api import AgentCatalogApi
from astraform_platform_client.api.agent_journeys_api import AgentJourneysApi
from astraform_platform_client.api.simulation_events_api import SimulationEventsApi
from astraform_platform_client.models.agent_definition_create_request import AgentDefinitionCreateRequest
from astraform_platform_client.models.agent_definition_publish_request import AgentDefinitionPublishRequest
from astraform_platform_client.models.agent_journey_launch_request import AgentJourneyLaunchRequest

client = ApiClient(Configuration(host="http://localhost:8484", retries=0))
catalog = AgentCatalogApi(client)
sample = json.loads(Path("sample-configuration.json").read_text())
capabilities = SimulationEventsApi(client).inspect_public_native_domain_capabilities(
    sample["launchRequestTemplate"]["domain"]["bindingKey"], _request_timeout=60.0)
print(capabilities.to_json())

draft = sample["catalogDraftRequest"]
draft["requestId"] = str(uuid4())
draft["agentKey"] = "marketing-customer-" + uuid4().hex[:12]
created = catalog.create_agent_definition(
    AgentDefinitionCreateRequest.from_dict(draft), _request_timeout=60.0)
definition = created.definition
revision = catalog.publish_agent_definition(
    definition.catalog_agent_id,
    AgentDefinitionPublishRequest(
        expected_draft_version=definition.draft_version,
        expected_draft_spec_digest=definition.draft_spec_digest),
    _request_timeout=60.0).revision

body = sample["launchRequestTemplate"]
body.update(requestId=str(uuid4()), catalogAgentId=str(definition.catalog_agent_id),
            catalogRevisionId=str(revision.catalog_revision_id),
            catalogDefinitionDigest=revision.catalog_definition_digest)
launch = AgentJourneyLaunchRequest.from_dict(body)
output = Path("run-" + str(launch.request_id))
output.mkdir()
(output / "launch-request.json").write_text(launch.to_json())
result = AgentJourneysApi(client).launch_agent_journey(launch, _request_timeout=60.0)
(output / "launch-response.json").write_text(result.to_json())
print("runId:", result.run_id, "saved to:", output)

The platform owns subsequent customer decisions, partner exchanges and logical time progression. This caller does not drive turns or advance a clock. If a launch response is lost, reload the saved launch-request.json into AgentJourneyLaunchRequest and resubmit that exact request, rather than rerunning the code that creates new IDs. Preserve omitted optional fields and JSON scalar types; do not fill in defaults during replay.

For the banking sample, use inspect_public_customer_mcp_capabilities(binding_key) before publication and copy the returned binding/tool revision and digest pins into the Catalog MCP grant, as its existing run-customer.py does. Its toolContexts bind partner-owned contextRef and subjectRef; native domain discovery alone does not grant customer access.

Inspect status and export evidence

Use the saved response for later reads. Substitute the directory printed above; this step does not create or relaunch a customer:

import json
from pathlib import Path
from astraform_platform_client import ApiClient, Configuration
from astraform_platform_client.api.simulation_runs_api import SimulationRunsApi

output = Path("run-<saved-launch-request-id>")
run_id = json.loads((output / "launch-response.json").read_text())["runId"]
client = ApiClient(Configuration(host="http://localhost:8484", retries=0))
runs = SimulationRunsApi(client)
print(runs.get_simulation_run(run_id, _request_timeout=60.0).to_json())
print(runs.inspect_simulation_run_ledger(run_id, limit=100, _request_timeout=60.0).to_json())

# Once evidence is ready, this returns HTTP 204 with no body.
runs.get_simulation_run_evidence_export_readiness(run_id, _request_timeout=60.0)
(output / "evidence.zip").write_bytes(
    runs.export_simulation_run_evidence(run_id, _request_timeout=60.0))

Readiness/export can return 409 while evidence is unavailable and 404 for an unknown run. Handle the SDK ApiException and retry the read when appropriate; do not launch a replacement run. Inspect terminal status, ledger and goal outcome together: successful transport or a valid customer response does not mean the goal succeeded. The 6 October Java/Python workflow verification had 11 valid first-attempt customer responses, but all three configured goals failed for recorded behavior/policy reasons. It establishes this bounded integration, not customer reliability or multi-year/1,000-customer capacity.

Build and contract provenance

Public API definitions are owned by remote-domain-contracts. contracts/platform-api.lock.json pins the platform-api bundle version, release asset and checksum. Setuptools resolves that dependency automatically into ignored .cache/platform-api/ and runs OpenAPI Generator 7.24.0 using pom.xml. It generates the existing experiment/customer API surface. No internal agent-runtime API or sibling platform checkout is an input.

The shared v1.2.0 contract release is published. Provider and platform specifications use the same version and tag, with separate checksum-pinned archives. Normal builds use those released inputs:

python -m build remote-domain-author-kit-python --outdir dist/author-kit
python -m pip install --force-reinstall dist/author-kit/astraform_remote_domain_author_kit-0.3.1-py3-none-any.whl
python -I scripts/smoke-platform-client.py

The resolver verifies the pinned SHA on downloads and cached archives. ASTRAFORM_PLATFORM_API_OFFLINE=true forbids downloads after the dependency has been resolved once. To test unpublished contract changes, explicitly supply ASTRAFORM_CONTRACT_ZIP and ASTRAFORM_PLATFORM_API_ZIP. Those inputs are kept separately, marked LOCAL_PRERELEASE, and never selected implicitly by later builds. CI and release require published inputs for both bundles.

Generated code is ignored under src/astraform_platform_client/. Wheel and source archive include it and the generated contracts/platform-api.snapshot.json plus release lock. Source archives build without Java, the contract repository or a network connection once Python build requirements are installed. Missing generated sources fail the build instead of producing a client-less author kit.

To update API definitions, change them in the contract repository, publish the shared release, and update both SDK contract locks to the same version and tag. No copied API tree is maintained in this SDK.

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 the remote-domain-contracts release 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.