#!/usr/bin/env python3
"""
Basic Usage Example for Amicus MCP Context Bus
This script demonstrates how to use the Amicus MCP context bus for
multi-agent coordination. It shows how to:
- Read the current shared state
- Register an agent
- Add messages to the message bus
- Update task status
- Add completed work entries
Usage:
python examples/basic_usage.py
"""
import sys
import time
from pathlib import Path
# Add the src directory to the Python path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from amicus.core import get_state_file, read_with_lock, write_with_lock
def main():
"""Demonstrate basic context bus operations."""
# Step 1: Read the current shared state
print("=" * 60)
print("Step 1: Reading shared state from context bus")
print("=" * 60)
state_file = get_state_file()
print(f"State file location: {state_file}\n")
state = read_with_lock(state_file)
if not state:
print("No existing state found. Initializing new state...")
state = {
"project_name": "example-project",
"start_time": time.time(),
"agents": {},
"tasks": [],
"messages": [],
"completed_work": []
}
else:
print(f"Project: {state.get('project_name', 'Unknown')}")
print(f"Active agents: {len(state.get('agents', {}))}")
print(f"Total tasks: {len(state.get('tasks', []))}")
print(f"Messages: {len(state.get('messages', []))}\n")
# Step 2: Register yourself as an agent
print("=" * 60)
print("Step 2: Registering as an agent")
print("=" * 60)
agent_id = "example_agent"
agent_info = {
"name": "Example Agent",
"role": "Demonstrator",
"model": "claude-sonnet-4-5",
"status": "active",
"joined_at": time.time()
}
state["agents"][agent_id] = agent_info
print(f"Registered agent: {agent_info['name']}")
print(f"Agent ID: {agent_id}")
print(f"Role: {agent_info['role']}\n")
# Step 3: Add a message to the message bus
print("=" * 60)
print("Step 3: Broadcasting a message")
print("=" * 60)
message = {
"from": agent_info['name'],
"type": "broadcast",
"message": "Hello! This is an example message from the basic usage script.",
"timestamp": time.time()
}
state["messages"].append(message)
print(f"Message sent: {message['message']}\n")
# Step 4: Create and update a task
print("=" * 60)
print("Step 4: Working with tasks")
print("=" * 60)
# Create a new task
new_task = {
"id": "example-task-1",
"title": "Example demonstration task",
"description": "This is a sample task created by the basic usage example",
"status": "in_progress",
"priority": "low",
"created_by": agent_info['name'],
"assigned_to": agent_info['name'],
"created_at": time.time(),
"started_at": time.time()
}
state["tasks"].append(new_task)
print(f"Created task: {new_task['title']}")
print(f"Task ID: {new_task['id']}")
print(f"Status: {new_task['status']}\n")
# Step 5: Mark task as completed and add to completed work
print("=" * 60)
print("Step 5: Completing a task")
print("=" * 60)
# Update task status
for task in state["tasks"]:
if task["id"] == "example-task-1":
task["status"] = "completed"
task["completed_at"] = time.time()
task["completion_details"] = "Successfully demonstrated basic usage patterns"
# Add to completed work log
completed_entry = {
"task_id": "example-task-1",
"completed_by": agent_info['name'],
"completed_at": time.time(),
"files_modified": ["examples/basic_usage.py"],
"summary": "Demonstrated core context bus functionality"
}
state["completed_work"].append(completed_entry)
print(f"Task marked as completed")
print(f"Completion details: {completed_entry['summary']}\n")
# Step 6: Write the updated state back
print("=" * 60)
print("Step 6: Writing updated state to context bus")
print("=" * 60)
write_with_lock(state_file, state)
print("State successfully updated!\n")
# Step 7: Read back and verify
print("=" * 60)
print("Step 7: Verifying the update")
print("=" * 60)
updated_state = read_with_lock(state_file)
print(f"Active agents: {len(updated_state.get('agents', {}))}")
print(f"Total tasks: {len(updated_state.get('tasks', []))}")
print(f"Completed tasks: {sum(1 for t in updated_state.get('tasks', []) if t.get('status') == 'completed')}")
print(f"Messages: {len(updated_state.get('messages', []))}")
print(f"Completed work entries: {len(updated_state.get('completed_work', []))}\n")
print("=" * 60)
print("Example completed successfully!")
print("=" * 60)
print("\nKey Takeaways:")
print("- Use read_with_lock() to safely read shared state")
print("- Use write_with_lock() to safely update shared state")
print("- Register agents in the 'agents' dictionary")
print("- Use 'messages' list for inter-agent communication")
print("- Track tasks in the 'tasks' list with status updates")
print("- Log completed work in 'completed_work' for auditability")
if __name__ == "__main__":
main()
