SDKs
Python
matter-sdk — Pydantic-modelled Matter SDK. Sync and async clients, httpx-backed transport, full type-stub coverage. Targets Python 3.11+.
Last updated
matter-sdk Beta
Pydantic models for every resource, sync and async clients, httpx-backed transport.
Strict typing via NewType-aliased IDs and Pydantic validators. Targets Python 3.11+.
TL;DR. pip install matter-sdk, construct with Matter(api_key=..., version=...),
call matter.entities.create(...). Mutations return Pydantic models with status="draft";
terminal outcomes arrive via webhook. Idempotency keys auto-generate. Errors raise
typed exceptions extending MatterError.
Installation
pip install matter-sdkuv add matter-sdkpoetry add matter-sdkrye add matter-sdk| Requirement | Version |
|---|---|
| Python | >= 3.11 |
| Pydantic | >= 2.5 (vendored as a dependency) |
httpx | >= 0.27 (default transport) |
| Type stubs | First-class py.typed marker — works with mypy, pyright, ty |
The package ships with full Pydantic v2 models. Every resource, every nested schema, every RFC 7807 problem object is a typed model — autocomplete in your IDE matches the OpenAPI spec exactly.
Authentication
The client requires api_key and version at construction. The SDK refuses to start
without version — there is no implicit "latest."
from matter import Matter
matter = Matter(
api_key=os.environ["MATTER_API_KEY"],
version="2026-05-01",
)For agent runtimes, pass a tok_ instead:
agent = Matter(
api_key="tok_4Kj2m8pQNq3KcAbc...",
version="2026-05-01",
)Async client:
from matter import AsyncMatter
async with AsyncMatter(api_key="...", version="2026-05-01") as matter:
entity = await matter.entities.create(jurisdiction="US-DE", legal_name="...")Optional construction kwargs:
| Kwarg | Type | Default | Description |
|---|---|---|---|
api_key | str | — | sk_live_, sk_test_, pk_live_, pk_test_, or tok_. |
version | str | — | Dated version, e.g. "2026-05-01". Required. |
base_url | str | https://api.mattermode.com/v1 | Override for staging or self-hosted. |
timeout | float | 30.0 | Per-request seconds before aborting. |
max_retries | int | 4 | 429 and 5xx only. |
transport | httpx.BaseTransport | default | Inject for testing or custom DNS. |
log_hook | Callable[[Request, Response], None] | None | Opt-in per-request log. |
First request
Form a Delaware C-Corp. The pending Pydantic model arrives in the response; the terminal
entity.active event arrives via webhook minutes later.
from matter import Matter
matter = Matter(api_key=os.environ["MATTER_API_KEY"], version="2026-05-01")
entity = matter.entities.create(
jurisdiction="US-DE",
legal_name="Waypoint Systems, Inc.",
founders=[
{"name": "Ada Lovelace", "email": "ada@waypoint.dev", "equity_pct": 60},
{"name": "Grace Hopper", "email": "grace@waypoint.dev", "equity_pct": 40},
],
)
print(entity.id) # ent_Nq3KcAbc — typed as EntityId
print(entity.status) # EntityStatus.DRAFT
print(entity.livemode) # False
print(entity.model_dump_json(indent=2))entity is a matter.models.Entity Pydantic model. All fields are typed; status is an
Enum; nested foreign keys use NewType aliases so EntityId cannot be passed where a
DocumentId is expected.
Idempotency
Every POST is idempotent on the server side for 24 hours. The SDK auto-generates a UUID
v4 per call. Override with idempotency_key= when retrying across process restarts:
grant = matter.grants.create(
entity="ent_Nq3KcAbc",
stakeholder="stk_4Kj2m8pQ",
share_class="cls_Common",
quantity=100_000,
idempotency_key="grant-2026-q2-ada-lovelace",
)Behavior:
- Caller-provided keys are passed verbatim. Up to 255 chars.
- Auto-generated keys are UUID v4, not stored. If you need durable retries, supply your own.
- Retries inside a single call (across
max_retries) reuse the same key. - Token-scoped: keys collide only within the credential that issued them.
Pagination
List endpoints return a synchronous (or async) iterator that walks every page transparently:
for entity in matter.entities.list():
if entity.status == EntityStatus.ACTIVE:
print(entity.legal_name)Async variant:
async for entity in matter.entities.list(): # AsyncMatter only
...For backpressure or large lists, walk pages manually:
cursor = None
while True:
page = matter.entities.list(limit=100, starting_after=cursor).page()
for e in page.data:
handle(e)
if not page.has_more:
break
cursor = page.next_cursorThe page envelope mirrors the API: Page[Entity](object="list", data=[...], has_more=..., next_cursor=...).
Errors
Every non-2xx becomes a typed exception. The class corresponds to the RFC 7807 type
URI — https://mattermode.com/errors/invalid_request becomes MatterValidationError.
from matter.errors import (
MatterError,
MatterValidationError,
MatterAuthorizationError,
MatterRateLimitError,
MatterIdempotencyReplayError,
MatterUpstreamUnavailableError,
)
try:
matter.entities.create(jurisdiction="US-XX")
except MatterValidationError as err:
# 400 — fix the request, do not retry
print(err.errors) # field-level details
except MatterRateLimitError as err:
# 429 — SDK already retried up to max_retries; surface to caller
print(f"retry after {err.retry_after}s")
except MatterError as err:
print(err.code, err.request_id, err.type)| Class | HTTP | RFC 7807 code |
|---|---|---|
MatterValidationError | 400 | invalid_request |
MatterAuthenticationError | 401 | authentication_required |
MatterAuthorizationError | 403 | authorization_required |
MatterNotFoundError | 404 | resource_not_found |
MatterStateConflictError | 409 | invalid_state_transition |
MatterIdempotencyReplayError | 409 | idempotency_key_replay |
MatterRateLimitError | 429 | rate_limit_exceeded |
MatterUpstreamUnavailableError | 503 | upstream_unavailable |
Full taxonomy: error handling.
Streaming
The events firehose is an async iterator wrapping the SSE endpoint
GET /v1/events/stream. The SDK handles reconnection, last-event-id replay, and
heartbeat ticks transparently.
async with AsyncMatter(api_key="...", version="2026-05-01") as matter:
async for event in matter.events.stream(
entities=["ent_Nq3KcAbc"],
types=["entity.state_changed", "filing.completed"],
):
print(event.type, event.sequence, event.data.id)For sync code, the streaming helper uses a background thread; prefer the async client in long-running services.
with matter.events.stream() as stream:
for event in stream:
handle(event)Strict per-entity ordering is preserved across reconnects via Last-Event-Id.
Webhooks
Verify signatures in your receiver. Use the raw request body — re-serialized JSON will not verify. The helper does a constant-time compare and validates the timestamp window (5 min default).
from flask import Flask, request, abort
from matter import Matter
from matter.errors import MatterSignatureError
matter = Matter(api_key="...", version="2026-05-01")
app = Flask(__name__)
@app.post("/webhooks/matter")
def receive():
try:
event = matter.webhooks.verify_signature(
payload=request.get_data(), # bytes — raw body
signature_header=request.headers["Matter-Signature"],
secret=os.environ["MATTER_WEBHOOK_SECRET"],
)
except MatterSignatureError:
abort(401)
handle(event)
return "", 204verify_signature returns a typed Event model on success and raises
MatterSignatureError on tampering, replay (>5 min skew), or secret mismatch.
Agent tokens
Create a tier-3 token. Tier-3 means the agent prepares mutations but a human signs the
Authorization resource before execution.
token = matter.tokens.create(
tier=3,
scopes=[{"allow": ["entities.read", "filings.create", "documents.write"]}],
principal={"human_id": "usr_4Kj2m8pQ", "agent_id": "agt_Nq3KcAbc"},
api_version="2026-05-01",
)
print(token.id) # tok_…
print(token.secret) # shown once — store securelyWhen a tier-3 agent attempts a write, the SDK raises MatterAuthorizationPendingError
carrying the Authorization resource ID:
from matter.errors import MatterAuthorizationPendingError
try:
agent.filings.create(entity="ent_Nq3KcAbc", form="DE-AnnualReport")
except MatterAuthorizationPendingError as err:
print(f"awaiting human signature: {err.authorization_id}")The dashboard or matter.authorizations.approve(auth_id) releases the action. See
agents for the full pattern.
Logging and observability
Opt in to a per-request log hook:
def log_hook(req, res):
print({
"method": req.method,
"path": req.url.path,
"status": res.status_code,
"request_id": res.headers.get("x-request-id"),
"duration_ms": res.elapsed.total_seconds() * 1000,
})
matter = Matter(
api_key="...",
version="2026-05-01",
log_hook=log_hook,
)The hook fires post-response (including failures). It does not see request bodies
unless you opt in via log_hook=LogHook(include_body=True, ...) — bodies may contain
PII.
For OpenTelemetry, use the httpx instrumentation directly — the SDK surfaces its
underlying client via matter._client (private; stable shape).
Resource catalog
Every API resource maps to an SDK namespace. Cross-link to the API reference for fields and constraints.
| SDK attribute | API resource | ID prefix |
|---|---|---|
matter.entities | Entity | ent_ |
matter.intents | Intent | int_ |
matter.incorporator_receipts | IncorporatorReceipt | rcp_ |
matter.stakeholders | Stakeholder | stk_ |
matter.equity_plans | EquityPlan | plan_ |
matter.share_classes | ShareClass | cls_ |
matter.share_ledger | ShareLedgerEntry | led_ |
matter.grants | Grant | grt_ |
matter.valuations | Valuation | val_ |
matter.documents | Document | doc_ |
matter.resolutions | Resolution | res_ |
matter.filings | Filing | flg_ |
matter.qualifications | Qualification | qal_ |
matter.registered_agents | RegisteredAgent | ra_ |
matter.tax_profiles | TaxProfile | tax_ |
matter.corporate_transactions | CorporateTransaction | ctx_ |
matter.tokens | Token | tok_ |
matter.authorizations | Authorization | auth_ |
matter.audit_entries | AuditEntry | aud_ |
matter.events | Event | evt_ |
matter.webhook_endpoints | WebhookEndpoint | whe_ |
matter.portfolios | Portfolio | pf_ |
matter.requests | Request | req_ |
matter.cap_table | CapTable view | (computed) |
matter.compliance | Compliance view | (computed) |
Migration notes
The SDK is pre-1.0. Until 1.0, breaking changes ship in minor versions and are called
out in /changelog. Pin a minor (matter-sdk~=0.7.0) and review the diff
on each bump.
Anticipated breaking changes for 1.0:
- Drop sync facade for endpoints that have async-only semantics (
events.stream). - Promote
matter._clienttomatter.clientonce the underlying transport contract stabilizes. - Move from positional arguments on
.create()calls to keyword-only (already the recommended pattern; will be enforced).
What's next
Error handling
Typed exception hierarchy, retry policy by class, side-by-side with Node and Go.
Webhooks
Flask, FastAPI, and Django receiver patterns for verified delivery.
Pagination
Sync iterators, async iterators, and manual cursor walks.
Agents
Tier-3 / tier-4 token creation and the Authorization pause pattern.