kaptanto
Python SDK for Kaptanto CDC — pydantic
ChangeEvent models and an httpx SSE streaming client.
Install
pip install kaptanto
# Optional LangChain StructuredTool helper:
pip install 'kaptanto[langchain]'
Models
from kaptanto import ChangeEvent, Operation
raw = {
"id": "01J5Z0000000000000000000A1",
"idempotency_key": "pg:public.orders:1:insert:0/1A000001",
"timestamp": "2026-06-15T12:00:00Z",
"source": "postgres://cdc@localhost:5432/shop",
"operation": "insert",
"table": "orders",
"key": {"id": 1},
"before": None,
"after": {"id": 1, "status": "pending"},
"metadata": {},
}
ev = ChangeEvent.model_validate(raw)
assert ev.operation is Operation.INSERT
assert ev.is_insert()
Field names match the Go event.ChangeEvent JSON tags exactly. Optional
ai_context carries opaque AI enrichment metadata when present.
Streaming
import asyncio
from kaptanto import KaptantoStream
async def main() -> None:
stream = KaptantoStream(
"http://localhost:7654/events",
consumer="my-svc",
token="...", # optional bearer token
tables=["orders"], # optional filter
operations=["insert", "update"],
)
try:
async for ev in stream:
print(ev.operation, ev.table, ev.after)
finally:
await stream.aclose()
asyncio.run(main())
Sync wrapper
from kaptanto import KaptantoStream
stream = KaptantoStream("http://localhost:7654/events", consumer="my-svc")
for ev in stream.iter_events():
print(ev.table, ev.operation)
The client reconnects with exponential backoff + jitter on disconnect, ignores
SSE comment pings, skips malformed frames with a warning, and resumes via the
stable consumer ID (server-side cursor).
LangChain
LangChain is an optional extra. Core pip install kaptanto never imports it.
Reactive agent pattern (preferred)
Wire the stream directly into any LangChain / LangGraph runnable:
from kaptanto import KaptantoStream
stream = KaptantoStream(
"http://localhost:7654/events",
consumer="orders-agent",
token="...",
tables=["orders"],
)
try:
async for ev in stream:
await agent.ainvoke({"input": ev.model_dump_json()})
finally:
await stream.aclose()
as_tool — poll recent events
For agents that pull CDC context on demand:
from kaptanto import KaptantoStream
from kaptanto.langchain import as_tool
stream = KaptantoStream("http://localhost:7654/events", consumer="tool-agent")
tool = as_tool(stream, max_events=20, timeout_s=2.0)
# bind `tool` into your agent; each invoke drains recent ChangeEvents as JSON
License
Apache-2.0
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file kaptanto-0.1.0.tar.gz.
File metadata
- Download URL: kaptanto-0.1.0.tar.gz
- Upload date:
- Size: 12.1 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f5afd7e38c023ff3bcaeea99f1650408fed62953792427a24d66b7563138d4c5
|
|
| MD5 |
1a7afa8eb4f1e0ab3d815b59f9a05aa9
|
|
| BLAKE2b-256 |
1742fdf0954c24a2d3a61db5ebf0c84e8626758536abe7532bf91fa5b2d23cad
|
Provenance
The following attestation bundles were made for kaptanto-0.1.0.tar.gz:
Publisher:
pypi-publish.yml on olucasandrade/kaptanto
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
kaptanto-0.1.0.tar.gz -
Subject digest:
f5afd7e38c023ff3bcaeea99f1650408fed62953792427a24d66b7563138d4c5 - Sigstore transparency entry: 2516153942
- Sigstore integration time:
-
Permalink:
olucasandrade/kaptanto@540d91107a4e41c0dbf6b3364a27edccd4b943f5 -
Branch / Tag:
refs/tags/python-v0.1.0 - Owner: https://github.com/olucasandrade
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
pypi-publish.yml@540d91107a4e41c0dbf6b3364a27edccd4b943f5 -
Trigger Event:
push
-
Statement type:
File details
Details for the file kaptanto-0.1.0-py3-none-any.whl.
File metadata
- Download URL: kaptanto-0.1.0-py3-none-any.whl
- Upload date:
- Size: 8.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
cd7535b235ca8ec3b700598816aa163007ec106b7005ef9ef465adf57663a46b
|
|
| MD5 |
0c0d2ce224f7c2d282d49ccbdc434dd3
|
|
| BLAKE2b-256 |
a41dcc52d0b7532f30e63ac5002351bf01d27432a2d2f448a529de3bf822b8c4
|
Provenance
The following attestation bundles were made for kaptanto-0.1.0-py3-none-any.whl:
Publisher:
pypi-publish.yml on olucasandrade/kaptanto
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
kaptanto-0.1.0-py3-none-any.whl -
Subject digest:
cd7535b235ca8ec3b700598816aa163007ec106b7005ef9ef465adf57663a46b - Sigstore transparency entry: 2516154019
- Sigstore integration time:
-
Permalink:
olucasandrade/kaptanto@540d91107a4e41c0dbf6b3364a27edccd4b943f5 -
Branch / Tag:
refs/tags/python-v0.1.0 - Owner: https://github.com/olucasandrade
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
pypi-publish.yml@540d91107a4e41c0dbf6b3364a27edccd4b943f5 -
Trigger Event:
push
-
Statement type:
