# Client Python

> Python client library for Danube. Use when generating Python producer/consumer code for testing.

- **Type:** Skill
- **Install:** `agentstack add skill-danube-messaging-danube-agent-skills-python`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [danube-messaging](https://agentstack.voostack.com/s/danube-messaging)
- **Installs:** 0
- **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills)
- **Latest version:** 0.1.0
- **License:** Apache-2.0
- **Upstream author:** [danube-messaging](https://github.com/danube-messaging)
- **Source:** https://github.com/danube-messaging/danube-agent-skills/tree/main/clients/python

## Install

```sh
agentstack add skill-danube-messaging-danube-agent-skills-python
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# Skill: Python Client — danube-client

## Prerequisites

- Python 3.9+ installed
- pip installed

## Installation

```bash
pip install danube-client
```

## Client Creation

```python
from danube import DanubeClientBuilder

async def main():
    client = await (
        DanubeClientBuilder()
        .service_url("http://127.0.0.1:6650")
        .build()
    )
```

## Producer

### Basic Producer

```python
producer = (
    client.new_producer()
    .with_topic("/default/my-topic")
    .with_name("my-producer")
    .build()
)

await producer.create()

message_id = await producer.send("Hello Danube!".encode())
print(f"Sent message ID: {message_id}")
```

### Send with Attributes

```python
attributes = {
    "source": "app-1",
    "priority": "high",
}

message_id = await producer.send(b"Important message", attributes)
```

### Partitioned Producer

```python
producer = (
    client.new_producer()
    .with_topic("/default/high-throughput")
    .with_name("partitioned-producer")
    .with_partitions(3)
    .build()
)

await producer.create()
```

### Reliable Dispatch Producer

```python
from danube import DispatchStrategy

producer = (
    client.new_producer()
    .with_topic("/default/critical-events")
    .with_name("reliable-producer")
    .with_dispatch_strategy(DispatchStrategy.RELIABLE)
    .build()
)

await producer.create()
```

### Schema-Linked Producer

```python
import json

producer = (
    client.new_producer()
    .with_topic("/default/events")
    .with_name("schema-producer")
    .with_schema_subject("event-schema")
    .build()
)

await producer.create()

event = json.dumps({"user_id": "user-123", "event": "login", "timestamp": 1234567890})
await producer.send(event.encode())
```

## Consumer

### Basic Consumer (Exclusive)

```python
from danube import SubType

consumer = (
    client.new_consumer()
    .with_topic("/default/my-topic")
    .with_consumer_name("my-consumer")
    .with_subscription("my-subscription")
    .with_subscription_type(SubType.EXCLUSIVE)
    .build()
)

await consumer.subscribe()

queue = await consumer.receive()

while True:
    message = await queue.get()
    payload = message.payload.decode()
    print(f"Received: {payload}")
    await consumer.ack(message)
```

### Subscription Types

```python
# Exclusive: single consumer, ordered
.with_subscription_type(SubType.EXCLUSIVE)

# Shared: load balanced across consumers
.with_subscription_type(SubType.SHARED)

# Failover: active/standby
.with_subscription_type(SubType.FAIL_OVER)

# Key-Shared: per-key ordering with parallelism
.with_subscription_type(SubType.KEY_SHARED)
```

### NACK with Retry

```python
while True:
    message = await queue.get()
    try:
        process(message)
        await consumer.ack(message)
    except Exception as e:
        await consumer.nack(message, delay_ms=1000, reason=f"failed: {e}")
```

### Partitioned Consumer

```python
# Consumer auto-discovers all partitions
consumer = (
    client.new_consumer()
    .with_topic("/default/my-topic")  # Parent topic name
    .with_consumer_name("partition-consumer")
    .with_subscription("partition-sub")
    .with_subscription_type(SubType.EXCLUSIVE)
    .build()
)

await consumer.subscribe()
queue = await consumer.receive()
# Automatically receives from all partitions
```

## Key-Shared

### Producer with Routing Keys

```python
# All "payment" messages go to the same consumer
await producer.send_with_key(b"Payment for #1001", None, "payment")

# "shipping" goes to (potentially) a different consumer
await producer.send_with_key(b"Order #1001 shipped", None, "shipping")
```

### Key-Shared Consumer

```python
consumer = (
    client.new_consumer()
    .with_topic("/default/orders")
    .with_consumer_name("worker_1")
    .with_subscription("orders_sub")
    .with_subscription_type(SubType.KEY_SHARED)
    .build()
)

await consumer.subscribe()
queue = await consumer.receive()

while True:
    message = await queue.get()
    key = message.routing_key if message.HasField("routing_key") else ""
    print(f"key={key} payload={message.payload.decode()}")
    await consumer.ack(message)
```

### Key Filtering

```python
# Only receives "payment" and "invoice" messages
consumer = (
    client.new_consumer()
    .with_topic("/default/orders")
    .with_consumer_name("payments_worker")
    .with_subscription("orders_filtered")
    .with_subscription_type(SubType.KEY_SHARED)
    .with_key_filter("payment")
    .with_key_filter("invoice")
    .build()
)

# Or set multiple at once:
# .with_key_filters(["payment", "invoice"])
```

Filter patterns use glob syntax: `"payment"` (exact), `"ship*"` (prefix), `"eu-west-?"` (single char wildcard).

## Schema Registry

### Create Schema Client

```python
schema_client = client.schema()
```

### Register JSON Schema

```python
import json
from danube import SchemaType

json_schema = json.dumps({
    "type": "object",
    "properties": {
        "user_id": {"type": "string"},
        "event": {"type": "string"},
        "timestamp": {"type": "integer"},
    },
    "required": ["user_id", "event", "timestamp"],
})

schema_id = await (
    schema_client.register_schema("user-events")
    .with_type(SchemaType.JSON_SCHEMA)
    .with_schema_data(json_schema.encode())
    .execute()
)
```

### Register Avro Schema

```python
avro_schema = json.dumps({
    "type": "record",
    "name": "UserEvent",
    "fields": [
        {"name": "user_id", "type": "string"},
        {"name": "event", "type": "string"},
        {"name": "timestamp", "type": "long"},
    ],
})

schema_id = await (
    schema_client.register_schema("user-events-avro")
    .with_type(SchemaType.AVRO)
    .with_schema_data(avro_schema.encode())
    .execute()
)
```

### Get / List / Check

```python
# Get latest
schema = await schema_client.get_latest_schema("user-events")
print(f"Schema ID: {schema.schema_id}, Version: {schema.version}")

# List versions
versions = await schema_client.list_versions("user-events")

# Check compatibility
is_compatible, errors = await schema_client.check_compatibility(
    "user-events", new_schema.encode(), SchemaType.JSON_SCHEMA, None
)
```

### Producer Schema Version Strategies

```python
# Latest (default) — always uses newest version
.with_schema_subject("user-events")

# Pinned to version 2
.with_schema_version("user-events", 2)

# Minimum version 2 (uses latest >= 2)
.with_schema_min_version("user-events", 2)
```

## Complete Example: Simple Producer & Consumer

```python
import asyncio
import json
from danube import DanubeClientBuilder, SubType

async def main():
    client = await DanubeClientBuilder().service_url("http://127.0.0.1:6650").build()

    # Producer
    producer = (
        client.new_producer()
        .with_topic("/default/test_topic")
        .with_name("test_producer")
        .build()
    )
    await producer.create()

    for i in range(5):
        msg = f"Message {i}"
        msg_id = await producer.send(msg.encode())
        print(f"Sent: {msg} (ID: {msg_id})")

    # Consumer
    consumer = (
        client.new_consumer()
        .with_topic("/default/test_topic")
        .with_consumer_name("test_consumer")
        .with_subscription("test_sub")
        .with_subscription_type(SubType.EXCLUSIVE)
        .build()
    )
    await consumer.subscribe()

    queue = await consumer.receive()
    while True:
        message = await queue.get()
        print(f"Received: {message.payload.decode()}")
        await consumer.ack(message)

if __name__ == "__main__":
    asyncio.run(main())
```

## Reference Examples

Working examples in the [danube-py repository](https://github.com/danube-messaging/danube-py/tree/main/examples):
- `simple_producer_consumer.py` — basic produce/consume
- `json_producer.py` / `json_consumer.py` — JSON schema
- `avro_producer.py` / `avro_consumer.py` — Avro schema
- `key_shared_producer.py` / `key_shared_consumer.py` — Key-Shared subscription
- `key_shared_filtered_consumer.py` — Key filtering
- `partitions_producer.py` / `partitions_consumer.py` — Partitioned topics
- `reliable_dispatch_producer.py` / `reliable_dispatch_consumer.py` — Reliable delivery
- `schema_evolution.py` — Schema versioning

## Troubleshooting

### PEP 668: "externally-managed-environment" error

On modern Linux (Debian 12+, Ubuntu 24+), `pip install` fails with PEP 668 error. Use a virtual environment:

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install danube-client
```

The AI should create/reuse a venv automatically when installing the Python client.

### "no partitions found" on consumer.subscribe()

The topic **must exist** before calling `consumer.subscribe()`. If the topic doesn't exist, the consumer will fail with `"no partitions found"`. Create the topic first:

```bash
danube-admin topics create /default/my-topic
# or with reliable delivery:
danube-admin topics create /default/my-topic --dispatch-strategy reliable
```

This applies to all client libraries, not just Python.

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [danube-messaging](https://github.com/danube-messaging)
- **Source:** [danube-messaging/danube-agent-skills](https://github.com/danube-messaging/danube-agent-skills)
- **License:** Apache-2.0

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-danube-messaging-danube-agent-skills-python
- Seller: https://agentstack.voostack.com/s/danube-messaging
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.