Multi-Agent Team with Shared Memory Pattern
This is a complete starter implementation for a multi-agent team using Letta’s shared memory blocks for coordination. Posted for Discord user darkprobe.
Architecture
- Coordinator: Task assignment, escalation handling, feedback synthesis
- Researcher: Information gathering, documentation
- Coder: Implementation, programming tasks
- Reviewer: Code review, quality checks
Model Recommendations
| Role | Model | Why |
|---|---|---|
| Coordinator | Claude Sonnet 4.5 | Best tool calling for orchestration |
| Researcher | Claude Sonnet 4.5 | Tool calling for web search |
| Coder | Claude Opus 4.5 | Complex implementation |
| Reviewer | Claude Haiku 4.5 | Fast, cost-effective |
Alternative: Use Gemini 3 Flash for Reviewer if budget-conscious.
Full Implementation
from letta_client import Letta
import json
import os
import asyncio
from datetime import datetime
from typing import Optional, Dict, List, Any
from dataclasses import dataclass, asdict
from enum import Enum
class TaskStatus(Enum):
PENDING = "pending"
ASSIGNED = "assigned"
IN_PROGRESS = "in_progress"
BLOCKED = "blocked"
REVIEW = "review"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
id: str
description: str
status: TaskStatus
assigned_to: Optional[str] = None
dependencies: List[str] = None
created_at: str = None
completed_at: Optional[str] = None
feedback_id: Optional[str] = None
def __post_init__(self):
if self.created_at is None:
self.created_at = datetime.now().isoformat()
if self.dependencies is None:
self.dependencies = []
class AgentTeam:
VERSION = "1.1.0"
def __init__(self, api_key: str, base_url: str = "https://api.letta.com"):
self.client = Letta(api_key=api_key, base_url=base_url)
self.team_id = f"team_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
def create_shared_blocks(self):
"""Create shared memory blocks for coordination."""
self.tasks_block = self.client.blocks.create(
label=f"{self.team_id}_tasks",
value=json.dumps({
"version": self.VERSION,
"tasks": {},
"completed": []
}),
limit=10000
)
self.requests_block = self.client.blocks.create(
label=f"{self.team_id}_requests",
value=json.dumps({"pending": [], "resolved": []}),
limit=5000
)
self.escalations_block = self.client.blocks.create(
label=f"{self.team_id}_escalations",
value=json.dumps({"active": [], "resolved": []}),
limit=3000
)
self.workspace_block = self.client.blocks.create(
label=f"{self.team_id}_workspace",
value="# Team Workspace\n\n",
limit=15000
)
self.feedback_block = self.client.blocks.create(
label=f"{self.team_id}_feedback",
value=json.dumps({
"learnings": [],
"task_feedback": {},
"common_mistakes": [],
"best_practices": []
}),
limit=10000
)
return {
"tasks": self.tasks_block.id,
"requests": self.requests_block.id,
"escalations": self.escalations_block.id,
"workspace": self.workspace_block.id,
"feedback": self.feedback_block.id
}
def create_agent(self, name: str, model: str, system_prompt: str,
skills: List[str] = None) -> str:
"""Create an agent with shared blocks attached."""
memory_blocks = [
{"label": f"{self.team_id}_tasks", "value": "", "limit": 10000},
{"label": f"{self.team_id}_requests", "value": "", "limit": 5000},
{"label": f"{self.team_id}_escalations", "value": "", "limit": 3000},
{"label": f"{self.team_id}_workspace", "value": "", "limit": 15000},
{"label": f"{self.team_id}_feedback", "value": "", "limit": 10000},
]
if skills:
skills_content = "\n".join([f"- {s}" for s in skills])
memory_blocks.append({
"label": "skills",
"value": f"# Available Skills\n{skills_content}",
"limit": 3000
})
agent = self.client.agents.create(
name=f"{self.team_id}_{name}",
model=model,
memory_blocks=memory_blocks,
system=system_prompt
)
return agent.id
def create_coordinator(self) -> str:
"""Create coordinator agent."""
system = f"""You are the coordinator for team {self.team_id}.
Your shared memory blocks:
- {self.team_id}_tasks: Task assignments and status
- {self.team_id}_requests: Pending inter-agent requests
- {self.team_id}_escalations: Issues requiring your attention
- {self.team_id}_workspace: Shared outputs and deliverables
- {self.team_id}_feedback: Team learnings and best practices
TASK ASSIGNMENT PROCESS:
1. When new task arrives, analyze requirements
2. Check agent availability in tasks block
3. Assign based on:
- Research tasks → researcher
- Implementation → coder
- Review tasks → reviewer
- Unclear scope → claim yourself, clarify, then delegate
4. Update tasks block with assignment
COORDINATION LOOP:
Every cycle, check in this order:
1. escalations block - resolve blockers first
2. requests block - route questions between agents
3. tasks block - check for completed tasks, assign new ones
FEEDBACK CAPTURE:
When task completes:
1. Read the completed work from workspace
2. Analyze what worked/didn't work
3. Update feedback block with learning
4. If pattern emerges, add to best_practices or common_mistakes
Available workers:
- researcher: Information gathering, analysis, API docs
- coder: Implementation, programming, debugging
- reviewer: Code review, quality checks, security audit"""
return self.create_agent(
name="coordinator",
model="anthropic/claude-sonnet-4.5",
system_prompt=system
)
def create_researcher(self) -> str:
"""Create researcher agent."""
system = f"""You are the researcher for team {self.team_id}.
Your shared memory blocks:
- {self.team_id}_tasks: Your assigned research tasks
- {self.team_id}_requests: Where coders/reviewers ask you questions
- {self.team_id}_escalations: Use this to ask coordinator for clarification
- {self.team_id}_workspace: Store research findings here
- {self.team_id}_feedback: Learn from past research tasks
BEFORE STARTING RESEARCH:
1. Check feedback block for "common_mistakes" related to research
2. Look for "best_practices" in research methodology
3. Review similar completed tasks in tasks block
DURING RESEARCH:
- Post findings to workspace with clear structure
- If coder needs clarification, use requests block
- If scope unclear, escalate to coordinator
FEEDBACK LOOP:
When research completes:
1. Document sources and methodology in workspace
2. Note any surprises or challenges
3. If approach was novel and worked, this becomes a learning"""
return self.create_agent(
name="researcher",
model="anthropic/claude-sonnet-4.5",
system_prompt=system,
skills=["web_search", "summarize", "document_analysis"]
)
def create_coder(self) -> str:
"""Create coder agent."""
system = f"""You are the coder for team {self.team_id}.
Your shared memory blocks:
- {self.team_id}_tasks: Your coding assignments
- {self.team_id}_requests: Where you ask researcher for info
- {self.team_id}_escalations: Use this to ask coordinator for clarification
- {self.team_id}_workspace: Store code and outputs here
- {self.team_id}_feedback: Learn from past coding tasks
BEFORE CODING:
1. Check feedback block for "common_mistakes" in coding patterns
2. Review "best_practices" for code style and patterns
3. Check if similar task exists in completed tasks
COLLABORATION PATTERNS:
- Need API docs? → requests block to researcher
- Stuck on approach? → escalations to coordinator
- Code ready for review? → Update task status, coordinator routes to reviewer
FEEDBACK LOOP:
When coding completes:
1. Document any "gotchas" encountered
2. Note if requirements were unclear (helps coordinator improve)
3. If you found a better pattern, add to learnings"""
return self.create_agent(
name="coder",
model="anthropic/claude-opus-4.5",
system_prompt=system,
skills=["read_file", "write_file", "edit_file", "bash", "git"]
)
def create_reviewer(self) -> str:
"""Create reviewer agent."""
system = f"""You are the reviewer for team {self.team_id}.
Your shared memory blocks:
- {self.team_id}_tasks: Your review assignments
- {self.team_id}_requests: Where you ask coders for clarification
- {self.team_id}_escalations: Use this to ask coordinator for help
- {self.team_id}_workspace: Read code here, post review comments
- {self.team_id}_feedback: Learn from past review patterns
BEFORE REVIEWING:
1. Check feedback block for recurring issues
2. Look for security patterns in common_mistakes
3. Review acceptance criteria in task description
REVIEW PROCESS:
- Code issues? → requests block to coder with specifics
- Architecture concerns? → escalations to coordinator
- Approved? → Update workspace with "APPROVED" and task status
FEEDBACK LOOP:
After review:
1. Document issue patterns you found
2. If coder repeatedly makes same mistake, flag for feedback
3. Add security/performance insights to best_practices"""
return self.create_agent(
name="reviewer",
model="anthropic/claude-haiku-4.5",
system_prompt=system,
skills=["read_file", "analyze_code", "security_audit"]
)
def assign_task(self, task: Task, team_config: Dict) -> bool:
"""Assign a task to an agent via coordinator."""
block = self.client.blocks.retrieve(team_config["blocks"]["tasks"])
tasks_data = json.loads(block.value)
tasks_data["tasks"][task.id] = asdict(task)
self.client.blocks.update(
block_id=team_config["blocks"]["tasks"],
value=json.dumps(tasks_data)
)
self.client.agents.messages.create(
agent_id=team_config["agents"]["coordinator"],
messages=[{
"role": "user",
"content": f"New task assigned: {task.id} to {task.assigned_to}\nDescription: {task.description}"
}],
streaming=False
)
return True
def coordination_cycle(self, team_config: Dict):
"""Run one coordination cycle."""
coordinator_id = team_config["agents"]["coordinator"]
response = self.client.agents.messages.create(
agent_id=coordinator_id,
messages=[{
"role": "user",
"content": "Run coordination cycle: check escalations, requests, and task status. Take actions as needed."
}],
streaming=False
)
return response
def capture_task_feedback(self, task_id: str, feedback: Dict, team_config: Dict):
"""Capture feedback after task completion."""
block = self.client.blocks.retrieve(team_config["blocks"]["feedback"])
feedback_data = json.loads(block.value)
feedback_data["task_feedback"][task_id] = {
"timestamp": datetime.now().isoformat(),
"what_worked": feedback.get("what_worked", ""),
"what_didnt": feedback.get("what_didnt", ""),
"suggestions": feedback.get("suggestions", ""),
"patterns_observed": feedback.get("patterns", [])
}
if feedback.get("add_to_best_practices"):
feedback_data["best_practices"].append({
"task_id": task_id,
"practice": feedback["add_to_best_practices"],
"timestamp": datetime.now().isoformat()
})
if feedback.get("add_to_common_mistakes"):
feedback_data["common_mistakes"].append({
"task_id": task_id,
"mistake": feedback["add_to_common_mistakes"],
"timestamp": datetime.now().isoformat()
})
self.client.blocks.update(
block_id=team_config["blocks"]["feedback"],
value=json.dumps(feedback_data)
)
def get_learnings(self, team_config: Dict) -> Dict:
"""Retrieve all captured learnings."""
block = self.client.blocks.retrieve(team_config["blocks"]["feedback"])
return json.loads(block.value)
def deploy_team(self) -> Dict:
"""Deploy full team and return agent IDs."""
blocks = self.create_shared_blocks()
coordinator_id = self.create_coordinator()
researcher_id = self.create_researcher()
coder_id = self.create_coder()
reviewer_id = self.create_reviewer()
config = {
"team_id": self.team_id,
"version": self.VERSION,
"blocks": blocks,
"agents": {
"coordinator": coordinator_id,
"researcher": researcher_id,
"coder": coder_id,
"reviewer": reviewer_id
}
}
with open(f"{self.team_id}.json", "w") as f:
json.dump(config, f, indent=2)
return config
# Usage Example
if __name__ == "__main__":
team = AgentTeam(api_key=os.getenv("LETTA_API_KEY"))
config = team.deploy_team()
print(f"Team deployed: {config['team_id']}")
# Create and assign tasks
task = Task(
id="research_001",
description="Research OAuth2 implementation patterns",
status=TaskStatus.ASSIGNED,
assigned_to="researcher"
)
team.assign_task(task, config)
# Run coordination
team.coordination_cycle(config)
Key Features
- Task Assignment: Updates tasks block + notifies coordinator
- Coordination Loop: Coordinator checks escalations → requests → task status
- Feedback Capture: Stores what_worked, what_didnt, best_practices, common_mistakes
- Continuous Improvement: Each task makes the team smarter for the next one
Notes
- A2A messaging tools are deprecated (Nov 2025). This uses shared memory + coordinator polling pattern.
- Shared blocks are prefixed with team_id to avoid collisions
- Agents are persistent - update models via
client.agents.modify()without recreating
Source: Discord discussion with darkprobe, Feb 17 2026