Official Python SDK for A2AMCP (Agent-to-Agent Model Context Protocol) - enabling seamless communication between AI agents working on parallel development tasks.
pip install a2amcp
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())
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")
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)
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()
Agents automatically handle:
# Automatic lifecycle management
async with Agent(project, "001", "feature", "description") as agent:
# Agent is registered and heartbeat is running
pass
# Agent is automatically unregistered
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
)
# 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()
# 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)
# 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()
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()
WAIT: Wait for the lock to be released (default)ABORT: Raise exception immediately on conflictQUEUE: Wait in line for the resourceNEGOTIATE: Query the lock owner and negotiate@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']}")
async with project.monitor() as monitor:
async for event in monitor.events():
print(f"Event: {event.type} - {event.data}")
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")
See the examples/ directory for complete examples:
Contributions are welcome! Please see CONTRIBUTING.md for guidelines.
MIT License - see LICENSE for details.
FAQs
Python SDK for A2AMCP - Agent-to-Agent communication via Model Context Protocol
We found that a2amcp-sdk demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
/Security News
A flawed sandbox in @nestjs/devtools-integration lets attackers run code on your machine via CSRF, leading to full Remote Code Execution (RCE).
Product
Customize license detection with Socket’s new license overlays: gain control, reduce noise, and handle edge cases with precision.
Product
Socket now supports Rust and Cargo, offering package search for all users and experimental SBOM generation for enterprise projects.