A2AMCP Python SDK

Official Python SDK for A2AMCP (Agent-to-Agent Model Context Protocol) - enabling seamless communication between AI agents working on parallel development tasks.

Features

• Simple Integration: Get started with just a few lines of code
• Automatic Lifecycle Management: Registration, heartbeats, and cleanup handled automatically
• Conflict Resolution: Built-in strategies for handling file conflicts
• Type Safety: Full type hints for better IDE support
• Async/Await: Modern async Python for efficient operation
• Prompt Generation: Intelligent prompt builder for optimal agent instructions

Installation

pip install a2amcp

Quick Start

Basic Agent

import asyncio
from a2amcp import A2AMCPClient, Project, Agent

async def main():
    # Initialize client and project
    client = A2AMCPClient("localhost:5000")
    project = Project(client, "my-project")
    
    # Create and run an agent
    async with Agent(
        project=project,
        task_id="001",
        branch="feature/auth",
        description="Build authentication"
    ) as agent:
        # Add todos
        todo_id = await agent.todos.add("Implement login", priority=1)
        
        # Work with files
        async with agent.files.coordinate("src/auth.py") as file:
            # File is locked
            print(f"Working on {file}")
        # File automatically released
        
        # Mark todo complete
        await agent.todos.complete(todo_id)

asyncio.run(main())

Orchestrator Spawning Agents

from a2amcp import A2AMCPClient, Project, TaskConfig, AgentSpawner

async def spawn_agents():
    client = A2AMCPClient("localhost:5000")
    project = Project(client, "my-app")
    
    tasks = [
        TaskConfig(
            task_id="001",
            branch="feature/auth",
            description="Build authentication",
            shared_interfaces=["User"]
        ),
        TaskConfig(
            task_id="002", 
            branch="feature/profile",
            description="Build user profiles",
            depends_on=["001"]
        )
    ]
    
    spawner = AgentSpawner(project)
    sessions = await spawner.spawn_multiple(tasks, "/path/to/worktrees")
    print(f"Spawned {len(sessions)} agents")

Agent with Message Handling

from a2amcp import Agent

agent = Agent(project, "003", "feature/api", "Build API")

@agent.handles("interface")
async def handle_interface_query(message):
    if "User" in message['content']:
        return "User has: id, email, password, role"

@agent.on("todo_completed") 
async def on_todo_completed(event):
    print(f"Completed: {event['todo']['text']}")

async with agent:
    while True:
        await agent.process_messages()
        await asyncio.sleep(5)

Core Concepts

Project Context

All operations happen within a project context:

project = Project(client, "project-id")

# Access managers
agents = await project.agents.list()
interfaces = await project.interfaces.list()
todos = await project.todos.get_all()

Agent Lifecycle

Agents automatically handle:

• Registration on startup
• Heartbeats every 30-45 seconds
• Message checking
• Unregistration on shutdown

# Automatic lifecycle management
async with Agent(project, "001", "feature", "description") as agent:
    # Agent is registered and heartbeat is running
    pass
# Agent is automatically unregistered

File Coordination

Prevent conflicts with built-in file locking:

# Simple coordination
async with agent.files.coordinate("src/models.py") as file:
    # File is locked
    pass
# File is released

# Advanced with conflict strategies
await agent.files.lock(
    "src/models.py",
    strategy=ConflictStrategy.WAIT,
    timeout=60
)

Inter-Agent Communication

# Query another agent
response = await agent.communication.query(
    "task-002",
    "interface", 
    "What fields does User have?"
)

# Broadcast to all
await agent.communication.broadcast(
    "info",
    "User model updated with new fields"
)

# Check messages
messages = await agent.communication.check_messages()

Shared Interfaces

# Register an interface
await project.interfaces.register(
    agent.session_name,
    "User",
    "interface User { id: string; email: string; }",
    "src/types/user.ts"
)

# Require an interface (waits if needed)
user = await project.interfaces.require("User", timeout=60)

Todo Management

# Add todos
todo1 = await agent.todos.add("Design schema", priority=1)
todo2 = await agent.todos.add("Write tests", priority=2)

# Update status
await agent.todos.start(todo1)
await agent.todos.complete(todo1)

# Check all todos in project
all_todos = await project.todos.get_all()

Prompt Generation

Generate optimal prompts for agents:

from a2amcp import PromptBuilder

prompt = PromptBuilder("project-id")\
    .with_task({
        "task_id": "001",
        "branch": "feature/auth",
        "description": "Build authentication",
        "depends_on": ["database"],
        "shared_interfaces": ["User", "Session"]
    })\
    .with_coordination_rules()\
    .with_error_recovery()\
    .add_instruction("Use bcrypt for passwords")\
    .build()

Advanced Features

Conflict Resolution Strategies

• WAIT: Wait for the lock to be released (default)
• ABORT: Raise exception immediately on conflict
• QUEUE: Wait in line for the resource
• NEGOTIATE: Query the lock owner and negotiate

Event Handling

@agent.on('file_conflict')
async def handle_conflict(event):
    print(f"Conflict on {event['file']}")

@agent.on('interface_registered')
async def handle_new_interface(event):
    print(f"New interface: {event['name']}")

Project Monitoring

async with project.monitor() as monitor:
    async for event in monitor.events():
        print(f"Event: {event.type} - {event.data}")

Error Handling

from a2amcp import ConflictError, TimeoutError

try:
    await agent.files.lock("src/models.py", strategy=ConflictStrategy.ABORT)
except ConflictError as e:
    print(f"File locked by {e.conflict.agent}")
except TimeoutError:
    print("Could not acquire lock in time")

Complete Examples

See the examples/ directory for complete examples:

1. Basic agent registration and communication
2. Agent with message handlers
3. Orchestrator spawning multiple agents
4. Conflict resolution strategies
5. Working with shared interfaces
6. Advanced prompt generation
7. Todo-driven development
8. Project monitoring dashboard

Requirements

• Python 3.8+
• A2AMCP server running (typically at localhost:5000)
• Docker (if using containerized A2AMCP server)

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.

Links

• A2AMCP Documentation
• API Reference
• Examples

Support

• GitHub Issues: github.com/webdevtodayjason/A2AMCP/issues
• Discord: Join our community