ringforge 0.1.0

Python SDK for the Ringforge agent mesh platform

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

pip install ringforge

Quick Start

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

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.

# 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.

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.

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.

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.

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.

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.

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 in the repository root.