diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml
new file mode 100644
index 0000000..495d9af
--- /dev/null
+++ b/.github/workflows/python.yml
@@ -0,0 +1,40 @@
+name: Python Client CI
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - 'clients/python/**'
+      - '.github/workflows/python.yml'
+  pull_request:
+    branches: [main]
+    paths:
+      - 'clients/python/**'
+      - '.github/workflows/python.yml'
+
+jobs:
+  test:
+    name: pytest (py${{ matrix.python-version }})
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: clients/python
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.10', '3.11', '3.12']
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+          cache-dependency-path: clients/python/pyproject.toml
+
+      - name: Install
+        run: pip install -e ".[test]"
+
+      - name: Run unit tests (live tests skip — no selfhost in CI)
+        run: pytest -v --tb=short
diff --git a/clients/python/.gitignore b/clients/python/.gitignore
new file mode 100644
index 0000000..3933934
--- /dev/null
+++ b/clients/python/.gitignore
@@ -0,0 +1,7 @@
+__pycache__/
+*.py[co]
+*.egg-info/
+build/
+dist/
+.venv/
+.pytest_cache/
diff --git a/clients/python/README.md b/clients/python/README.md
new file mode 100644
index 0000000..8d4db1e
--- /dev/null
+++ b/clients/python/README.md
@@ -0,0 +1,52 @@
+# contexto (Python)
+
+Python client for the self-hosted Contexto memory API. Targets the OSS memory engine exposed by `packages/openrouter` (default port 4010).
+
+> This client is for the **self-hosted** stack only. The hosted `api.getcontexto.com` API has a different surface (mindmap-based) and is not supported here.
+
+## Install
+
+```bash
+pip install "contexto @ git+https://github.com/amiller/contexto.git#subdirectory=clients/python"
+```
+
+Or for local dev:
+
+```bash
+pip install -e clients/python/
+```
+
+## Usage
+
+```python
+from contexto import ContextoClient
+
+ctx = ContextoClient(base_url="http://localhost:4010")
+
+# Register an agent (idempotent on the slug)
+ctx.register_agent("hermes", name="Hermes")
+
+# Ingest a turn
+ctx.ingest(
+    messages=[
+        {"role": "user", "content": "My data center has 12 server racks."},
+        {"role": "assistant", "content": "Got it — 12 racks."},
+    ],
+    agent="hermes",
+    user_id="alex",
+)
+
+# Multi-sector cognitive memory query
+result = ctx.search("how many racks", agent="hermes", user_id="alex")
+# -> {"workingMemory": [...], "perSector": {...}, "agentId": "hermes"}
+
+# Or a formatted block ready for prompt injection
+block = ctx.get_context_for_turn("how many racks", agent="hermes", user_id="alex")
+```
+
+## Endpoints used
+
+- `POST /v1/agents` — register an agent
+- `POST /v1/ingest` — ingest `{messages, agent, userId}`
+- `POST /v1/search` — query `{query, agent, userId}`
+- `GET  /v1/summary` — recent memories
diff --git a/clients/python/contexto/__init__.py b/clients/python/contexto/__init__.py
new file mode 100644
index 0000000..10dcda9
--- /dev/null
+++ b/clients/python/contexto/__init__.py
@@ -0,0 +1,4 @@
+from .client import ContextoClient, DEFAULT_BASE
+
+__all__ = ["ContextoClient", "DEFAULT_BASE"]
+__version__ = "0.1.0"
diff --git a/clients/python/contexto/client.py b/clients/python/contexto/client.py
new file mode 100644
index 0000000..5f5b3f5
--- /dev/null
+++ b/clients/python/contexto/client.py
@@ -0,0 +1,149 @@
+"""Contexto self-hosted API client.
+
+Targets the OSS memory engine exposed by `packages/openrouter` (default
+http://localhost:4010), which embeds the memory router. Endpoints:
+  POST /v1/agents     register an agent
+  POST /v1/ingest     ingest a turn (LLM-extracts components into 3 sectors)
+  POST /v1/search     query the working memory across sectors
+  GET  /v1/summary    list recent memories
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+import httpx
+
+DEFAULT_BASE = "http://localhost:4010"
+
+
+def _raise_with_body(resp: httpx.Response) -> None:
+    """raise_for_status() but with the server's response body in the message.
+
+    httpx's default error string is just status + URL — not enough to debug
+    extraction or embedding failures from the selfhost server.
+    """
+    if resp.is_success:
+        return
+    body = resp.text[:1000]
+    raise httpx.HTTPStatusError(
+        f"{resp.status_code} {resp.reason_phrase} from {resp.request.url}: {body}",
+        request=resp.request,
+        response=resp,
+    )
+
+
+class ContextoClient:
+    """Client for the self-hosted Contexto memory API."""
+
+    def __init__(self, base_url: str = DEFAULT_BASE, timeout: float = 300.0):
+        # Default 300s: /v1/ingest runs LLM extraction + embeddings server-side.
+        # Empirical: ~48s for a 4K-char hermes turn; ~120-180s for the long-tail
+        # outliers (longer turns + a slow Gemini response). 300s catches the
+        # 99th percentile without papering over a real hang.
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+        self.headers = {"Content-Type": "application/json"}
+
+    def list_agents(self) -> dict:
+        resp = httpx.get(f"{self.base_url}/v1/agents", headers=self.headers, timeout=self.timeout)
+        _raise_with_body(resp)
+        return resp.json()
+
+    def register_agent(
+        self,
+        agent_id: str,
+        name: Optional[str] = None,
+        soul: Optional[str] = None,
+        relevance_prompt: Optional[str] = None,
+    ) -> dict:
+        """Register a new agent. Idempotent on the slug."""
+        body: dict = {"id": agent_id}
+        if name is not None:
+            body["name"] = name
+        if soul is not None:
+            body["soul"] = soul
+        if relevance_prompt is not None:
+            body["relevancePrompt"] = relevance_prompt
+        resp = httpx.post(
+            f"{self.base_url}/v1/agents",
+            headers=self.headers,
+            json=body,
+            timeout=self.timeout,
+        )
+        _raise_with_body(resp)
+        return resp.json()
+
+    def ingest(
+        self,
+        messages: list[dict],
+        agent: str,
+        user_id: Optional[str] = None,
+    ) -> dict:
+        """Ingest a sequence of {role, content} messages.
+
+        At least one user-role message with non-empty content is required.
+        Server runs LLM extraction and stores components across episodic,
+        semantic, and procedural sectors.
+        """
+        body: dict = {"messages": messages, "agent": agent}
+        if user_id:
+            body["userId"] = user_id
+        resp = httpx.post(
+            f"{self.base_url}/v1/ingest",
+            headers=self.headers,
+            json=body,
+            timeout=self.timeout,
+        )
+        _raise_with_body(resp)
+        return resp.json()
+
+    def search(self, query: str, agent: str, user_id: Optional[str] = None) -> dict:
+        """Multi-sector cognitive memory query.
+
+        Returns {workingMemory, perSector, agentId}. workingMemory is the
+        cap-limited cross-sector top hits; perSector is the per-sector lists.
+        """
+        body: dict = {"query": query, "agent": agent}
+        if user_id:
+            body["userId"] = user_id
+        resp = httpx.post(
+            f"{self.base_url}/v1/search",
+            headers=self.headers,
+            json=body,
+            timeout=self.timeout,
+        )
+        _raise_with_body(resp)
+        return resp.json()
+
+    def summary(self, agent: str, limit: int = 50, user_id: Optional[str] = None) -> dict:
+        params: dict = {"agent": agent, "limit": limit}
+        if user_id:
+            params["userId"] = user_id
+        resp = httpx.get(
+            f"{self.base_url}/v1/summary",
+            headers=self.headers,
+            params=params,
+            timeout=self.timeout,
+        )
+        _raise_with_body(resp)
+        return resp.json()
+
+    def get_context_for_turn(
+        self,
+        query: str,
+        agent: str,
+        user_id: Optional[str] = None,
+        max_results: int = 5,
+    ) -> str:
+        """Run a search and format workingMemory hits as a prompt-ready block."""
+        result = self.search(query, agent=agent, user_id=user_id)
+        items = (result or {}).get("workingMemory") or []
+        parts = []
+        for it in items[:max_results]:
+            sector = it.get("sector", "?")
+            score = it.get("score", 0)
+            content = it.get("content", "").strip()
+            if content:
+                parts.append(f"[{sector} | score: {score:.2f}] {content}")
+        return "\n---\n".join(parts)
diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml
new file mode 100644
index 0000000..b10cdbc
--- /dev/null
+++ b/clients/python/pyproject.toml
@@ -0,0 +1,33 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "contexto"
+version = "0.1.0"
+description = "Python client for the self-hosted Contexto memory engine."
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "MIT" }
+authors = [{ name = "ekailabs" }]
+dependencies = [
+  "httpx>=0.25",
+]
+
+[project.optional-dependencies]
+test = [
+  "pytest>=7",
+]
+
+[project.urls]
+Homepage = "https://github.com/ekailabs/contexto"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["contexto*"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+markers = [
+  "live: requires a running Contexto selfhost (skips if CONTEXTO_BASE_URL is unreachable)",
+]
diff --git a/clients/python/tests/__init__.py b/clients/python/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/clients/python/tests/test_client.py b/clients/python/tests/test_client.py
new file mode 100644
index 0000000..d9ede05
--- /dev/null
+++ b/clients/python/tests/test_client.py
@@ -0,0 +1,237 @@
+"""Unit tests for ContextoClient — mocked HTTP, no selfhost required.
+
+These are the tests CI will run. They cover request body shapes, default
+behavior, and error-message formatting. A separate `test_live.py` covers
+the integration path against a real selfhost.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+import pytest
+
+from contexto import ContextoClient
+from contexto.client import _raise_with_body, DEFAULT_BASE
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+class _Recorder:
+    """Captures the last httpx.post / httpx.get call so tests can assert on it."""
+
+    def __init__(self):
+        self.calls: list[dict[str, Any]] = []
+        self.next_response: httpx.Response | None = None
+
+    def make_post(self):
+        def _post(url, *, headers=None, json=None, timeout=None, **kwargs):
+            self.calls.append({"method": "POST", "url": url, "headers": headers,
+                               "json": json, "timeout": timeout})
+            return self.next_response or _ok({})
+        return _post
+
+    def make_get(self):
+        def _get(url, *, headers=None, params=None, timeout=None, **kwargs):
+            self.calls.append({"method": "GET", "url": url, "headers": headers,
+                               "params": params, "timeout": timeout})
+            return self.next_response or _ok({})
+        return _get
+
+
+def _ok(body: dict, status: int = 200) -> httpx.Response:
+    return httpx.Response(status, json=body, request=httpx.Request("POST", "http://x/"))
+
+
+def _err(status: int, body: str) -> httpx.Response:
+    return httpx.Response(status, content=body.encode(),
+                          request=httpx.Request("POST", "http://x/"))
+
+
+@pytest.fixture
+def recorder(monkeypatch) -> _Recorder:
+    rec = _Recorder()
+    monkeypatch.setattr("contexto.client.httpx.post", rec.make_post())
+    monkeypatch.setattr("contexto.client.httpx.get", rec.make_get())
+    return rec
+
+
+# ---------------------------------------------------------------------------
+# Construction defaults
+# ---------------------------------------------------------------------------
+
+def test_default_base_is_localhost():
+    c = ContextoClient()
+    assert c.base_url == "http://localhost:4010"
+    assert DEFAULT_BASE == "http://localhost:4010"
+
+
+def test_default_timeout_is_generous():
+    """Ingest runs server-side LLM extraction (Gemini ~10-180s depending on
+    turn size). Default httpx timeout (5s) and even a 120s ceiling miss the
+    long tail. 300s catches the 99th percentile."""
+    assert ContextoClient().timeout >= 300.0
+
+
+def test_base_url_trailing_slash_stripped():
+    assert ContextoClient(base_url="http://h:4010/").base_url == "http://h:4010"
+
+
+def test_no_auth_headers_for_selfhost():
+    """Selfhost has no auth — Authorization header must not be set."""
+    headers = ContextoClient().headers
+    assert "Content-Type" in headers
+    assert "Authorization" not in headers
+
+
+# ---------------------------------------------------------------------------
+# Request body shapes
+# ---------------------------------------------------------------------------
+
+def test_register_agent_minimal_body(recorder):
+    recorder.next_response = _ok({"agent": {"id": "h"}})
+    ContextoClient().register_agent("hermes")
+    body = recorder.calls[-1]["json"]
+    assert body == {"id": "hermes"}
+
+
+def test_register_agent_includes_optional_fields(recorder):
+    recorder.next_response = _ok({"agent": {"id": "h"}})
+    ContextoClient().register_agent("hermes", name="Hermes", soul="be wise",
+                                    relevance_prompt="filter noise")
+    body = recorder.calls[-1]["json"]
+    assert body == {"id": "hermes", "name": "Hermes", "soul": "be wise",
+                    "relevancePrompt": "filter noise"}
+
+
+def test_ingest_body_no_user_id(recorder):
+    recorder.next_response = _ok({"stored": 1, "ids": ["a"], "agent": "h"})
+    msgs = [{"role": "user", "content": "hi"}]
+    ContextoClient().ingest(messages=msgs, agent="hermes")
+    body = recorder.calls[-1]["json"]
+    assert body == {"messages": msgs, "agent": "hermes"}
+    assert "userId" not in body
+
+
+def test_ingest_body_with_user_id(recorder):
+    recorder.next_response = _ok({"stored": 1, "ids": ["a"], "agent": "h"})
+    msgs = [{"role": "user", "content": "hi"}]
+    ContextoClient().ingest(messages=msgs, agent="hermes", user_id="alex")
+    assert recorder.calls[-1]["json"]["userId"] == "alex"
+
+
+def test_ingest_url_and_timeout(recorder):
+    recorder.next_response = _ok({"stored": 0, "ids": [], "agent": "h"})
+    c = ContextoClient(base_url="http://h:4010", timeout=200)
+    c.ingest(messages=[{"role": "user", "content": "x"}], agent="hermes")
+    call = recorder.calls[-1]
+    assert call["url"] == "http://h:4010/v1/ingest"
+    assert call["timeout"] == 200
+
+
+def test_search_body(recorder):
+    recorder.next_response = _ok({"workingMemory": [], "perSector": {}, "agentId": "h"})
+    ContextoClient().search("rate limiter", agent="hermes", user_id="alex")
+    assert recorder.calls[-1]["json"] == {
+        "query": "rate limiter", "agent": "hermes", "userId": "alex",
+    }
+
+
+def test_summary_uses_query_params(recorder):
+    recorder.next_response = _ok({"recent": [], "summary": []})
+    ContextoClient().summary(agent="hermes", limit=20, user_id="alex")
+    call = recorder.calls[-1]
+    assert call["method"] == "GET"
+    assert call["url"].endswith("/v1/summary")
+    assert call["params"] == {"agent": "hermes", "limit": 20, "userId": "alex"}
+
+
+# ---------------------------------------------------------------------------
+# Response handling
+# ---------------------------------------------------------------------------
+
+def test_get_context_for_turn_formats_working_memory(recorder):
+    recorder.next_response = _ok({
+        "workingMemory": [
+            {"sector": "semantic", "score": 0.71, "content": "12 server racks"},
+            {"sector": "episodic", "score": 0.65, "content": "rack 7 humidity"},
+        ],
+        "perSector": {}, "agentId": "h",
+    })
+    block = ContextoClient().get_context_for_turn("racks", agent="hermes")
+    assert "[semantic | score: 0.71] 12 server racks" in block
+    assert "[episodic | score: 0.65] rack 7 humidity" in block
+    assert "\n---\n" in block
+
+
+def test_get_context_for_turn_empty_returns_empty_string(recorder):
+    recorder.next_response = _ok({"workingMemory": [], "perSector": {}, "agentId": "h"})
+    assert ContextoClient().get_context_for_turn("anything", agent="h") == ""
+
+
+def test_get_context_for_turn_skips_blank_content(recorder):
+    recorder.next_response = _ok({
+        "workingMemory": [
+            {"sector": "semantic", "score": 0.71, "content": ""},
+            {"sector": "semantic", "score": 0.65, "content": "   "},
+            {"sector": "episodic", "score": 0.5, "content": "real one"},
+        ],
+        "perSector": {}, "agentId": "h",
+    })
+    block = ContextoClient().get_context_for_turn("anything", agent="h")
+    # Only the real one survives; the empties are filtered out.
+    assert "real one" in block
+    assert block.count("[") == 1
+
+
+def test_get_context_for_turn_caps_at_max_results(recorder):
+    recorder.next_response = _ok({
+        "workingMemory": [{"sector": "s", "score": 0.5, "content": f"item{i}"}
+                          for i in range(10)],
+        "perSector": {}, "agentId": "h",
+    })
+    block = ContextoClient().get_context_for_turn("q", agent="h", max_results=3)
+    assert block.count("[") == 3
+
+
+# ---------------------------------------------------------------------------
+# Error surfacing
+# ---------------------------------------------------------------------------
+
+def test_raise_with_body_passes_on_success():
+    """No-op for 2xx — the helper must not raise on success."""
+    _raise_with_body(_ok({"ok": True}))
+
+
+def test_raise_with_body_includes_status_and_body():
+    resp = _err(500, '{"error":"gemini extract failed: 429 quota exceeded"}')
+    with pytest.raises(httpx.HTTPStatusError) as excinfo:
+        _raise_with_body(resp)
+    msg = str(excinfo.value)
+    assert "500" in msg
+    assert "gemini extract failed" in msg
+    assert "quota exceeded" in msg
+
+
+def test_raise_with_body_truncates_huge_bodies():
+    resp = _err(500, "x" * 10_000)
+    with pytest.raises(httpx.HTTPStatusError) as excinfo:
+        _raise_with_body(resp)
+    # Body capped at 1000 chars in the message.
+    assert len(str(excinfo.value)) < 1500
+
+
+def test_ingest_propagates_server_error_with_body(recorder):
+    recorder.next_response = _err(500, '{"error":"extraction failed: bad json"}')
+    with pytest.raises(httpx.HTTPStatusError) as excinfo:
+        ContextoClient().ingest(messages=[{"role": "user", "content": "x"}], agent="h")
+    assert "extraction failed" in str(excinfo.value)
+
+
+def test_ingest_returns_parsed_json_on_success(recorder):
+    recorder.next_response = _ok({"stored": 3, "ids": ["a", "b", "c"], "agent": "h"})
+    out = ContextoClient().ingest(messages=[{"role": "user", "content": "x"}], agent="h")
+    assert out == {"stored": 3, "ids": ["a", "b", "c"], "agent": "h"}
diff --git a/clients/python/tests/test_live.py b/clients/python/tests/test_live.py
new file mode 100644
index 0000000..1abe568
--- /dev/null
+++ b/clients/python/tests/test_live.py
@@ -0,0 +1,96 @@
+"""Live integration test — exercises a real running Contexto selfhost.
+
+Skips automatically if no selfhost is reachable. Run locally with the stack
+up (`docker run ... contexto-selfhost`) to verify the client against the
+real wire format.
+
+In CI, this test is skipped unless the workflow brings up a stack.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import uuid
+
+import httpx
+import pytest
+
+from contexto import ContextoClient
+
+
+_BASE = os.environ.get("CONTEXTO_BASE_URL", "http://localhost:4010")
+
+
+def _selfhost_alive() -> bool:
+    try:
+        httpx.get(f"{_BASE}/v1/agents", timeout=2.0).raise_for_status()
+        return True
+    except Exception:
+        return False
+
+
+pytestmark = [
+    pytest.mark.live,
+    pytest.mark.skipif(
+        not _selfhost_alive(),
+        reason=f"Contexto selfhost not reachable at {_BASE}",
+    ),
+]
+
+
+@pytest.fixture
+def client() -> ContextoClient:
+    return ContextoClient(base_url=_BASE)
+
+
+@pytest.fixture
+def slug() -> str:
+    return f"pytest-{uuid.uuid4().hex[:8]}"
+
+
+def test_register_then_ingest_then_search(client: ContextoClient, slug: str):
+    client.register_agent(slug, name=slug)
+
+    # Ingest one turn with a distinctive fact.
+    fact = (
+        "The matrix-greeter project tracks rate-limit clock skew with the "
+        "histogram metric gateway_rate_limit_skew_ms."
+    )
+    r = client.ingest(
+        messages=[
+            {"role": "user", "content": fact},
+            {"role": "assistant", "content": "Got it — tracked via the histogram."},
+        ],
+        agent=slug,
+        user_id="pytest",
+    )
+    assert r["stored"] >= 1
+    assert r["agent"] == slug
+
+    # Give the server a moment to index.
+    time.sleep(2)
+
+    # Search recovers something semantically related.
+    result = client.search("how do we monitor clock skew", agent=slug, user_id="pytest")
+    wm = result.get("workingMemory") or []
+    assert wm, "expected at least one working-memory hit after ingest"
+
+    # And the formatted block matches the same content.
+    block = client.get_context_for_turn("clock skew", agent=slug, user_id="pytest")
+    assert block, "get_context_for_turn should return a non-empty block"
+    assert "score:" in block and "[" in block
+
+
+def test_summary_lists_recent(client: ContextoClient, slug: str):
+    client.register_agent(slug, name=slug)
+    client.ingest(
+        messages=[
+            {"role": "user", "content": "remember octarine is the user's favorite color"},
+            {"role": "assistant", "content": "Noted."},
+        ],
+        agent=slug, user_id="pytest",
+    )
+    time.sleep(2)
+    summary = client.summary(agent=slug, limit=20, user_id="pytest")
+    assert "recent" in summary or "summary" in summary