midplane 0.1.2

Official Python client for the MidPlane backend's API-agent surface.

midplane — Python client for MidPlane

Official Python client for the MidPlane backend's API-agent surface. Thin enough to drop into any script; opinionated where it matters (DataField envelopes, AtomRef wire shape, error mapping, proposal drafts).

Status: 0.1.0 alpha. API may shift; pin a patch version.

Install

pip install midplane

Requires Python ≥ 3.10. Runtime deps: httpx, pydantic ≥ 2.6.

Quick start

from midplane import MidplaneClient

with MidplaneClient.from_env() as client:        # reads MIDPLANE_API_TOKEN + MIDPLANE_API_URL
    me = client.whoami()
    print(me.atom_id, me.name)

    atom = client.atom(me.atom_id)
    atom.refresh()
    print(atom.name, atom.system_tags)

    for neighbor in atom.neighbors(depth=2).neighbors:
        print(neighbor.id, neighbor.name)

Set the env vars first:

export MIDPLANE_API_URL="https://midplane.cloud/api"
export MIDPLANE_API_TOKEN="mspsa_live_…_…"   # mint via UI; see the API guide

Bearer tokens

The token starts with mspsa_live_ for historical reasons (legacy namespace prefix; existing tokens and leak-scanners depend on it). The client doesn't mint tokens — that's an owner-side operation through the UI or install_api_agent. Once you have a token, set MIDPLANE_API_TOKEN and call MidplaneClient.from_env().

Authoritative reference for the underlying API: docs/api-agent-usage.md in the MidPlane repo.

What's in the box

from midplane import (
    MidplaneClient,       # sync client
    AsyncMidplaneClient,  # awaitable sibling
    AtomHandle,           # convenience wrapper around one atom
    ProposalDraft,        # context manager for incremental proposals

    # Typed responses (Pydantic v2)
    WhoAmI, Atom, HubSummary, HubAttached, SearchHit,
    NeighborsResult, NeighborIdsResult,
    Proposal, ProposalSubmitResult, ProposalDraftSummary,
    HealthReport, HealthCheck,

    # Errors — match on type; .raw carries the full backend payload
    MidplaneError,
    AuthError, PermissionError, NotFoundError,
    ValidationError, RateLimitError, TransportError,
)

Common workflows

Connectivity self-check

report = client.health()
if not report.ok:
    print(report.summary())   # human-readable: endpoint, auth, identity
    raise SystemExit(1)

health() never raises — it runs three progressively-deeper checks (reachability, token, whoami resolves) and returns a structured report with per-check latency and remediation hints. Useful in startup scripts and CI smoke tests. Use report.raise_for_status() if you'd rather get an exception on first failure.

Read an atom + walk

me = client.whoami()
hub = client.get_hub()                    # HubSummary, not a full Atom
print(hub.hub_id, hub.linked_agent_count, hub.attached_resource_count)

for n in client.get_neighbors_full(hub.hub_id, depth=2).neighbors:
    print(n.id, n.name, n.system_tags)

get_hub() returns a HubSummary (discoverability index); follow hub.hub_id into get_atom_full(...) if you need the full atom.

Write a local-data field

client.atom(atom_id).replace_field(
    "cap/wiki_article/inputs",
    [{"schema_version": 1, "body": {"type": "doc", "content": [...]}}],
)

No DataField envelopes. Pass raw Python values; the client never wraps. On reads, the client auto-unwraps the storage-layer envelopes ({"String": …}, {"Number": …}, etc.) so you get plain Python out.

Incremental proposal

with client.proposal_draft("Wire up new wiki atom") as draft:
    new_atom = draft.create_atom(name="Test Article", creator=parent_id)
    draft.install_capability(new_atom, "wiki_article")
    draft.set_field(
        new_atom,
        "cap/wiki_article/inputs",
        {"schema_version": 1, "body": {"type": "doc", "content": [...]}},
    )

# Submitted on context exit; the resulting Proposal is on draft.result.
print(draft.result.id, draft.result.proposal.status)

The helpers wrap atom refs into the backend's AtomRef enum shape ({"kind": "existing", "id": "<uuid>"} for an existing UUID, {"kind": "placeholder", "id": <int>} for a placeholder bound by a prior create_atom in the same draft). Pass bare strings or ints — the helpers handle the rest.

Exiting via exception discards the draft instead. To detach the draft from Python's lifecycle (let it live server-side past this block), open with auto_submit=False and call draft.detach() / .submit() / .discard() explicitly.

One-shot proposal (no draft)

from midplane.drafts import atom_ref

client.proposal_submit(
    title="Patch a field",
    operations=[{
        "op": "set_local_field",
        "atom": atom_ref(some_atom_id),
        "key": "demo/note",
        "value": "hello",
    }],
)

The draft helpers (draft.set_field(...) etc.) wrap refs for you; when you build ops by hand, call atom_ref(...) so the shape matches the backend's AtomRef enum.

Async

import asyncio
from midplane import AsyncMidplaneClient

async def main() -> None:
    async with AsyncMidplaneClient.from_env() as client:
        me = await client.whoami()
        print(me.atom_id)

asyncio.run(main())

Surface is method-for-method identical to the sync client, including client.health(). Drafts + AtomHandle helpers are sync-only in 0.1 (they call the sync client under the hood); use the raw await client.call(op, payload) escape hatch if you need them in async code.

Error handling

from midplane import PermissionError, NotFoundError

try:
    atom = client.get_atom_full(some_id)
except NotFoundError as e:
    print("missing:", e.atom_id)
except PermissionError as e:
    print("needed bits:", e.required_names)
    print("had bits:", e.granted_names)

PermissionError carries the full check/checks payload from the backend; .raw always has the full response dict for further drilling.

Calling an op the client doesn't wrap

result = client.call("some_new_dispatcher_op", {"foo": "bar"})

call(op, payload) is the escape hatch. It does not unwrap DataField envelopes — if you want unwrapped values from a wrapped response, run midplane.datafield.unwrap_local_data(...) on the appropriate sub-dict.

Gotchas

These mirror the API guide's Gotchas section; restated here because they're load-bearing for Python integrators.

• DataField envelopes (repeated for emphasis): never wrap. The client always speaks raw Python on writes; reads come back unwrapped. Adding {"String": …} envelopes by hand is the single most common integration mistake.
• AtomRef wrapping in proposal ops. Any field that takes an atom in a proposal op (atom, src, dest, creator) is the AtomRef enum on the wire, not a bare UUID string. Use the draft helpers, or call midplane.drafts.atom_ref(...) when hand-building ops.
• sub_id ≠ key. remove_local_data needs the per-entry UUID returned by reads, not the field key. Get it from a prior get_local_data(..., raw=True) (raw entries carry sub_id).
• Replace-all vs. append. replace_field / update_local_data discards every existing entry under the key and writes your new list. set_field / add_local_data appends one entry and preserves siblings.
• get_hub() is a summary. Returns HubSummary (id + counts + attached resources), not a full Atom. Follow hub.hub_id into get_atom_full for the full record.
• Argon2 cost. Bearer auth runs Argon2 verify on every request (~10ms per call). Use bulk_get_atom_full(ids) over many single get_atom_full(id) calls when you're processing > 50 atoms at once.
• Cascade after connect_atoms. Auto-org-enroll + policy reconciliation fires after connect_atoms returns. If you follow up with a read expecting cascade-derived state, wait or poll. See the API guide's §9e and §11 (Concurrency).
• Restart safety. Tokens survive backend restarts; in-flight requests don't. Most ops are idempotent (update_local_data, connect_atoms); create_atom, proposal_submit, install_api_agent are not. A RetryPolicy is on the 0.2 roadmap.

Smoke test

examples/feature_tour.py exercises nearly every public method against a live backend. Each section is isolated — a missing permission produces a soft skip, not a crash — so it doubles as a permissions self-audit for a freshly-minted agent token:

export MIDPLANE_API_URL="https://midplane.cloud/api"
export MIDPLANE_API_TOKEN="mspsa_live_…"
python examples/feature_tour.py

The simpler examples/counter.py is a minimal read-increment-write loop — start here if you just want to confirm the library is wired up.

Development

# Set up a venv and install in editable mode
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

# Run unit tests (no live backend needed; uses httpx.MockTransport)
PYTHONPATH=src python -m unittest discover -s tests/unit -v

# Type-check
pyright

# Lint
ruff check src tests

44 unit tests cover envelope unwrap, error classification, request envelope shape, response parsing, health-report diagnostics, proposal-op wire shapes, and the no-wrap write-side contract.

License

MIT.