agentkit-py

AgentKit

A lightweight Python framework for building multi-agent systems with session management, message routing, state persistence, and lock management.

Installation

pip install agentkit-py

Features

• Session Management - Track agent sessions with ownership, channels, and metadata
• Message Routing - Send messages between agents with pub/sub support
• State Persistence - Key-value store with namespaces and atomic operations
• Lock Management - Distributed locking with TTL and cross-process support

Quick Start

from agentkit import SessionManager, MessageBus, StateStore, LockManager
from pathlib import Path

# Create managers with persistence
data_dir = Path("./agent_data")
sessions = SessionManager(data_dir / "sessions")
messages = MessageBus(data_dir / "messages")
state = StateStore(data_dir / "state.json")
locks = LockManager(data_dir / "locks")

# Create a session
session = sessions.create(
    agent_id="agent-1",
    channel="telegram",
    owner="user-123",
)

# Send messages between agents
messages.send(
    from_agent="agent-1",
    to_agent="agent-2", 
    content="Hello from agent 1!",
)

# Receive messages
msg = messages.receive("agent-2")
if msg:
    print(f"Got: {msg.content}")

# Persist state
state.set("last_run", "2024-01-15", namespace="agent-1")
state.increment("run_count", namespace="agent-1")

# Lock shared resources
with locks.hold("shared-resource", owner="agent-1", ttl_seconds=30) as lock:
    if lock:
        # Do exclusive work
        pass

Session Management

from agentkit import SessionManager

sessions = SessionManager("./sessions")

# Create session
session = sessions.create(
    agent_id="my-agent",
    channel="slack",
    owner="user-456",
    metadata={"workspace": "acme"}
)

# Find sessions
active = sessions.find(agent_id="my-agent", status="active")

# Update session
session.metadata["last_action"] = "processed_message"
sessions.update(session)

# Cleanup stale sessions (inactive > 1 hour)
cleaned = sessions.cleanup_stale(max_age_seconds=3600)

Message Routing

from agentkit import MessageBus

bus = MessageBus("./messages")

# Send a message
msg = bus.send(
    from_agent="coordinator",
    to_agent="worker-1",
    content="Process this task",
    metadata={"task_id": "abc123"}
)

# Subscribe to messages
def handle_message(msg):
    print(f"Received: {msg.content}")

bus.subscribe("worker-1", handle_message)

# Broadcast to all agents
bus.broadcast(
    from_agent="coordinator",
    content="System maintenance in 5 minutes"
)

# Check pending messages
count = bus.pending_count("worker-1")

State Persistence

from agentkit import StateStore

store = StateStore("./state.json")

# Basic operations
store.set("config", {"timeout": 30}, namespace="agent-1")
config = store.get("config", namespace="agent-1")

# Atomic increment
count = store.increment("requests", namespace="agent-1")

# Append to list with max length
store.append("recent_errors", "timeout error", 
             namespace="agent-1", max_length=100)

# Get with metadata
info = store.get_with_metadata("config", namespace="agent-1")
# {'value': {...}, 'updated_at': 1705334400.0}

Lock Management

from agentkit import LockManager

locks = LockManager("./locks")

# Acquire lock with TTL
lock = locks.acquire(
    resource="database-connection",
    owner="agent-1",
    ttl_seconds=60
)

if lock:
    try:
        # Do exclusive work
        pass
    finally:
        locks.release("database-connection", "agent-1")

# Context manager (auto-release)
with locks.hold("resource", "agent-1", ttl_seconds=30) as lock:
    if lock:
        # Work with lock held
        pass

# Wait for lock
lock = locks.acquire(
    resource="busy-resource",
    owner="agent-1",
    wait=True,
    wait_timeout=10.0
)

# Force release stuck lock
locks.force_release("stuck-resource")

In-Memory Mode

All components work without persistence (in-memory only):

# No storage_dir = in-memory only
sessions = SessionManager()  
messages = MessageBus()
state = StateStore()
locks = LockManager()

For AI Agents

See SKILL.md for agent-optimized documentation with command patterns and examples.

License

MIT