feat: add schema registry helpers to Python SDK

Expose schema upsert/get client methods with tests and quickstart coverage so Track C parity includes the schemas family in GA clients.

Made-with: Cursor

Author: geobon

README.md

@@ -81,6 +81,17 @@ with AxmeClient(config) as client:
         idempotency_key="media-finalize-001",
     )
     print(finalized["status"])
+    schema = client.upsert_schema(
+        {
+            "semantic_type": "axme.calendar.schedule.v1",
+            "schema_json": {"type": "object", "required": ["date"], "properties": {"date": {"type": "string"}}},
+            "compatibility_mode": "strict",
+        },
+        idempotency_key="schema-upsert-001",
+    )
+    print(schema["schema"]["schema_hash"])
+    schema_get = client.get_schema("axme.calendar.schedule.v1")
+    print(schema_get["schema"]["semantic_type"])
     subscription = client.upsert_webhook_subscription(
         {
             "callback_url": "https://integrator.example/webhooks/axme",

axme_sdk/client.py

@@ -233,6 +233,25 @@ def finalize_media_upload(
             retryable=idempotency_key is not None,
         )
 
+    def upsert_schema(
+        self,
+        payload: dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+        trace_id: str | None = None,
+    ) -> dict[str, Any]:
+        return self._request_json(
+            "POST",
+            "/v1/schemas",
+            json_body=payload,
+            idempotency_key=idempotency_key,
+            trace_id=trace_id,
+            retryable=idempotency_key is not None,
+        )
+
+    def get_schema(self, semantic_type: str, *, trace_id: str | None = None) -> dict[str, Any]:
+        return self._request_json("GET", f"/v1/schemas/{semantic_type}", trace_id=trace_id, retryable=True)
+
     def upsert_webhook_subscription(
         self,
         payload: dict[str, Any],

tests/test_client.py

@@ -461,6 +461,71 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert client.finalize_media_upload(payload, idempotency_key="media-finalize-1")["status"] == "ready"
 
 
+def test_upsert_schema_success() -> None:
+    semantic_type = "axme.calendar.schedule.v1"
+    payload = {
+        "semantic_type": semantic_type,
+        "schema_json": {"type": "object", "required": ["date"], "properties": {"date": {"type": "string"}}},
+        "compatibility_mode": "strict",
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/v1/schemas"
+        assert request.headers["idempotency-key"] == "schema-upsert-1"
+        body = json.loads(request.read().decode("utf-8"))
+        assert body == payload
+        return httpx.Response(
+            200,
+            json={
+                "ok": True,
+                "schema": {
+                    "semantic_type": semantic_type,
+                    "schema_ref": f"schema://{semantic_type}",
+                    "schema_hash": "a" * 64,
+                    "compatibility_mode": "strict",
+                    "scope": "tenant",
+                    "owner_agent": "agent://owner",
+                    "active": True,
+                    "created_at": "2026-02-28T00:00:00Z",
+                    "updated_at": "2026-02-28T00:00:01Z",
+                },
+            },
+        )
+
+    client = _client(handler)
+    assert client.upsert_schema(payload, idempotency_key="schema-upsert-1")["schema"]["semantic_type"] == semantic_type
+
+
+def test_get_schema_success() -> None:
+    semantic_type = "axme.calendar.schedule.v1"
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "GET"
+        assert request.url.path == f"/v1/schemas/{semantic_type}"
+        return httpx.Response(
+            200,
+            json={
+                "ok": True,
+                "schema": {
+                    "semantic_type": semantic_type,
+                    "schema_ref": f"schema://{semantic_type}",
+                    "schema_hash": "b" * 64,
+                    "compatibility_mode": "strict",
+                    "scope": "tenant",
+                    "owner_agent": "agent://owner",
+                    "active": True,
+                    "schema_json": {"type": "object", "properties": {"date": {"type": "string"}}},
+                    "created_at": "2026-02-28T00:00:00Z",
+                    "updated_at": "2026-02-28T00:00:01Z",
+                },
+            },
+        )
+
+    client = _client(handler)
+    assert client.get_schema(semantic_type)["schema"]["semantic_type"] == semantic_type
+
+
 @pytest.mark.parametrize(
     ("status_code", "expected_exception"),
     [