agentic-ai-engineering/skills/kits/multi-agent-orch/examples/example-agent-team.py

125 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""
Example: Multi-Agent Team Configuration.
This script demonstrates how to define and run a multi-agent team
using the patterns from the Multi-Agent Orchestration skill kit.
Usage:
python example-agent-team.py
"""
import json
# ── Team Definition ──────────────────────────────────────────────
TEAM_CONFIG = {
"orchestrator": {
"name": "build-lead",
"model": "claude-sonnet-4",
"role": "Orchestrator — delegates work, never executes directly",
"max_turns": 25,
"agents": ["frontend-dev", "backend-dev", "reviewer"],
},
"frontend-dev": {
"name": "frontend-dev",
"model": "claude-haiku-3.5",
"role": "Builds UI components per spec",
"max_turns": 20,
"domain": {
"paths": ["src/frontend/"],
"allow_read": True,
"allow_write": True,
"allow_delete": False,
},
},
"backend-dev": {
"name": "backend-dev",
"model": "claude-sonnet-4",
"role": "Builds API endpoints and database queries",
"max_turns": 20,
"domain": {
"paths": ["src/api/", "src/db/"],
"allow_read": True,
"allow_write": True,
"allow_delete": False,
},
},
"reviewer": {
"name": "reviewer",
"model": "claude-sonnet-4",
"role": "Reviews code for quality, security, and style",
"max_turns": 15,
"domain": {
"paths": ["src/"],
"allow_read": True,
"allow_write": False,
"allow_delete": False,
},
"read_only": True,
},
}
# ── Chain Definition ─────────────────────────────────────────────
CHAIN_CONFIG = {
"name": "feature-pipeline",
"steps": [
{"agent": "planner", "input": "$ORIGINAL", "output": "plan.md"},
{"agent": "builder", "input": "plan.md", "output": "src/"},
{"agent": "reviewer", "input": "src/", "output": "review.md"},
{"agent": "verifier", "input": "src/", "output": "verified.md"},
],
}
# ── Domain Locking Check ─────────────────────────────────────────
def check_domain_access(agent_name: str, file_path: str) -> bool:
"""Check if an agent can access a file based on domain rules."""
agent = TEAM_CONFIG.get(agent_name)
if not agent:
return False
domain = agent.get("domain", {})
paths = domain.get("paths", [])
return any(file_path.startswith(p) for p in paths)
# ── Test ─────────────────────────────────────────────────────────
def test_team_config():
print(" Multi-Agent Team Configuration\n")
print(" Agent Roster:")
for name, config in TEAM_CONFIG.items():
role = config["role"][:60]
domain = config.get("domain", {}).get("paths", ["(no domain)"])
model = config["model"]
print(f" {name:20s} | {model:20s} | {domain[0]}")
print("\n Domain Locking Tests:")
tests = [
("frontend-dev", "src/frontend/Button.tsx", True),
("frontend-dev", "src/api/routes.ts", False),
("reviewer", "src/frontend/Button.tsx", True),
("reviewer", "src/frontend/", True), # read-only is OK
("backend-dev", "src/api/routes.ts", True),
("backend-dev", "src/frontend/Button.tsx", False),
]
for agent, path, expected in tests:
result = check_domain_access(agent, path)
status = "" if result == expected else ""
assert result == expected, f"FAIL: {agent} access to {path}"
print(f" {status} {agent:20s} can{'not ' if not expected else ' '}access {path}")
print("\n Chain Pipeline:")
for i, step in enumerate(CHAIN_CONFIG["steps"]):
print(f" Step {i+1}: {step['agent']:12s}{step['output']}")
print("\n All checks passed.\n")
if __name__ == "__main__":
test_team_config()