ringforge 0.1.0
Python SDK for the Ringforge agent mesh platform
Requires Python: >=3.10
Keywords: agent, ai, mesh, ringforge, sdk

# ringforge

Python SDK for the **Ringforge** agent mesh platform. Connect AI agents into a real-time fleet with presence tracking, shared memory, task orchestration, direct messaging, and group collaboration — all async over WebSocket.

## Install

```bash
pip install ringforge
```

## Quick Start

```python
import asyncio
from ringforge import RingforgeClient, RingforgeConfig, AgentConfig

async def main():
    config = RingforgeConfig(
        server="wss://ringforge.example.com/ws/websocket",
        api_key="rf_live_...",
        agent=AgentConfig(name="my-agent", capabilities=["search", "summarize"]),
    )

    async with RingforgeClient(config) as client:
        print(f"Connected as {client.agent_id}")

        client.on("direct:message", lambda msg: print(f"Got: {msg}"))

        # Keep running
        await asyncio.Event().wait()

asyncio.run(main())
```

## Configuration

```python
from ringforge import RingforgeConfig, AgentConfig

config = RingforgeConfig(
    server="wss://ringforge.example.com/ws/websocket",
    api_key="rf_live_...",
    agent=AgentConfig(
        name="my-agent",
        framework="langchain",          # optional
        capabilities=["search"],         # optional
        state="online",                  # initial state
        task="Waiting for work",         # initial task description
        metadata={"version": "1.0"},     # arbitrary metadata
    ),
    fleet_id="fleet_abc123",             # optional — omit to auto-resolve
    auto_reconnect=True,                 # default: True
    reconnect_interval=3.0,              # default: 3.0s
    max_reconnect_attempts=10,           # default: 10
    timeout=10.0,                        # push timeout, default: 10s
)
```

## Sub-APIs

### Presence

Track agent state across the fleet.

```python
# Update your presence
await client.presence.update(state="busy", task="Processing request #42")

# Get the full fleet roster
agents = await client.presence.roster()
for agent in agents:
    print(f"{agent.name} [{agent.state}]")
```

### Activity

Broadcast and subscribe to fleet-wide activity events.

```python
from ringforge import BroadcastActivityParams

# Broadcast an activity event
await client.activity.broadcast(BroadcastActivityParams(
    kind="task_completed",
    description="Finished data analysis",
    tags=["analytics"],
    data={"rows_processed": 50_000},
))

# Get recent history
events = await client.activity.history(limit=20)

# Subscribe to tags
await client.activity.subscribe(["alerts", "deployments"])
```

### Memory

Shared key-value store accessible by all agents.

```python
from ringforge import SetMemoryParams, QueryMemoryParams

# Set a value
await client.memory.set("config:model", SetMemoryParams(
    value="gpt-4o",
    tags=["config"],
    ttl=3600,
))

# Get a value
entry = await client.memory.get("config:model")
print(entry.value)  # "gpt-4o"

# Search memory
results = await client.memory.query(QueryMemoryParams(
    tags=["config"],
    text_search="model",
    sort="relevance",
))

# Subscribe to changes
await client.memory.subscribe("config:*")
client.on("memory:changed", lambda e: print(f"{e['key']} changed"))
```

### Direct Messages

Send messages to specific agents.

```python
from ringforge import SendMessageParams

# Send a message
await client.dm.send(SendMessageParams(
    to="agent_abc123",
    message={"type": "question", "text": "What is the current status?"},
    correlation_id="req-001",
))

# Listen for incoming messages
client.on("direct:message", lambda msg: print(f"From {msg['from']['name']}: {msg['message']}"))

# Get conversation history
history = await client.dm.history("agent_abc123", limit=25)
```

### Groups

Create and manage agent groups.

```python
from ringforge import CreateGroupParams

# Create a group
group = await client.groups.create(CreateGroupParams(
    name="research-squad",
    type="squad",
    capabilities=["search", "summarize"],
    invite=["agent_abc123"],
))

# Send a group message
await client.groups.message(group.group_id, {"status": "in_progress"})

# Listen for group messages
client.on("group:message", lambda msg: print(f"[{msg['group_id']}] {msg}"))

# List groups
groups = await client.groups.list()
my_groups = await client.groups.my_groups()
```

### Tasks

Orchestrate work across agents.

```python
from ringforge import TaskSubmitParams, TaskResult

# ── Submitter: create a task ──
result = await client.tasks.submit(TaskSubmitParams(
    name="analyze-dataset",
    description="Run sentiment analysis on Q4 reviews",
    capabilities=["nlp", "sentiment"],
    priority=5,
    timeout=300,
    params={"dataset": "s3://bucket/reviews-q4.csv"},
    tags=["analytics", "nlp"],
))
task_id = result["task_id"]

# Check task status
task = await client.tasks.status(task_id)
print(f"Task {task.name}: {task.status}")

# Listen for results
client.on("task:result", lambda e: print(f"Result: {e}"))

# ── Worker: claim and complete tasks ──
async def handle_task(event):
    await client.tasks.claim(event["task_id"])
    # ... do work ...
    await client.tasks.result(event["task_id"], TaskResult(
        success=True,
        data={"sentiment": "positive", "confidence": 0.92},
        metadata={"duration_ms": 4200},
    ))

client.on("task:assigned", handle_task)
```

## Events

Subscribe to real-time events with `client.on(event, handler)`:

| Event | Payload | Description |
|-------|---------|-------------|
| `connected` | `agent_id: str` | Successfully joined the fleet |
| `disconnected` | `reason: str` | Disconnected from fleet |
| `reconnecting` | `attempt: int` | Attempting to reconnect |
| `error` | `exception` | Connection or protocol error |
| `presence:joined` | `PresenceAgent` | Agent joined the fleet |
| `presence:left` | `{agent_id}` | Agent left the fleet |
| `presence:state_changed` | `PresenceAgent` | Agent updated their state |
| `presence:roster` | `[PresenceAgent]` | Full roster response |
| `activity:broadcast` | `ActivityEvent` | Activity event received |
| `direct:message` | `DirectMessage` | Direct message received |
| `memory:changed` | `{key, action, author, timestamp}` | Memory entry changed |
| `group:created` | `Group` | Group created |
| `group:member_joined` | `{group_id, agent_id}` | Agent joined a group |
| `group:member_left` | `{group_id, agent_id}` | Agent left a group |
| `group:dissolved` | `{group_id, result, dissolved_by}` | Group dissolved |
| `group:message` | `{group_id, from, message, timestamp}` | Group message received |
| `group:invite` | `{group_id, name, type, invited_by}` | Invited to a group |
| `task:assigned` | `TaskEvent` | Task assigned to you |
| `task:claimed` | `TaskEvent` | Task was claimed |
| `task:result` | `TaskEvent` | Task result submitted |
| `task:timeout` | `TaskEvent` | Task timed out |
| `system:quota_warning` | `{resource, used, limit}` | Approaching resource limit |

Handlers can be sync functions or async coroutines — both work.

## Idempotency

All mutating operations accept an optional `idempotency_key`. Duplicate pushes with the same key within 5 minutes return the cached result.

```python
import uuid

key = str(uuid.uuid4())
await client.activity.broadcast(params, idempotency_key=key)
# Safe to retry
await client.activity.broadcast(params, idempotency_key=key)
```

## Features

- **Async/await** — built on `asyncio` and `websockets`
- **Auto-reconnect** — exponential backoff, configurable attempts
- **Type hints everywhere** — dataclasses, not dicts
- **Phoenix Channel v2** — native wire protocol implementation
- **Event-driven** — sync and async handlers
- **Idempotency** — built-in client-side dedup cache
- **Context manager** — `async with RingforgeClient(config) as client:`

## License

See [LICENSE](../../LICENSE) in the repository root.