{"id":588,"date":"2026-03-21T02:09:10","date_gmt":"2026-03-20T18:09:10","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=588"},"modified":"2026-03-21T02:09:10","modified_gmt":"2026-03-20T18:09:10","slug":"a-coding-implementation-showcasing-clawteams-multi-agent-swarm-orchestration-with-openai-function-calling","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=588","title":{"rendered":"A Coding Implementation Showcasing ClawTeam\u2019s Multi-Agent Swarm Orchestration with OpenAI Function Calling"},"content":{"rendered":"

In this comprehensive tutorial, we present the core architecture of ClawTeam<\/strong><\/a>, an open-source Agent Swarm Intelligence framework developed by HKUDS. We implement the fundamental concepts that make ClawTeam powerful: a leader agent that decomposes complex goals into sub-tasks, specialized worker agents that execute those tasks autonomously, a shared task board with automatic dependency resolution, and an inter-agent messaging system that enables real-time coordination. We designed this tutorial to run seamlessly in Colab, requiring only an OpenAI API key, so anyone can experience multi-agent orchestration without setting up local infrastructure like tmux, git worktrees, or filesystem-based message queues that the original ClawTeam CLI requires.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import subprocess\nimport sys\n\n\ndef install_packages():\n   packages = [\"openai\", \"rich\"]\n   for pkg in packages:\n       subprocess.check_call(\n           [sys.executable, \"-m\", \"pip\", \"install\", \"-q\", pkg],\n           stdout=subprocess.DEVNULL,\n           stderr=subprocess.DEVNULL,\n       )\n   print(\"\"\u2705\" All packages installed.\")\n\n\ninstall_packages()\n\n\nimport os\nimport getpass\n\n\ntry:\n   from google.colab import userdata\n   OPENAI_API_KEY = userdata.get(\"OPENAI_API_KEY\")\n   if not OPENAI_API_KEY:\n       raise ValueError(\"Key not set in Colab secrets\")\n   print(\"\"\ud83d\udd11\" API key loaded from Colab Secrets.\")\nexcept Exception:\n   OPENAI_API_KEY = getpass.getpass(\"\"\ud83d\udd11\" Enter your OpenAI API key: \")\n   print(\"\"\ud83d\udd11\" API key received (hidden).\")\n\n\nos.environ[\"OPENAI_API_KEY\"] = OPENAI_API_KEY\n\n\nimport json\nimport uuid\nimport time\nimport threading\nfrom enum import Enum\nfrom datetime import datetime\nfrom dataclasses import dataclass, field\nfrom typing import Optional\nfrom collections import defaultdict\n\n\nfrom openai import OpenAI\nfrom rich.console import Console\nfrom rich.table import Table\nfrom rich.panel import Panel\nfrom rich.tree import Tree\nfrom rich.live import Live\nfrom rich.layout import Layout\nfrom rich.text import Text\n\n\nconsole = Console()\nclient = OpenAI(api_key=OPENAI_API_KEY)\n\n\nMODEL = \"gpt-4o-mini\"<\/code><\/pre>\n<\/div>\n<\/div>\n
We begin by installing the required packages, OpenAI and rich, and securely collecting the OpenAI API key through either Colab Secrets or terminal input. We then import all the libraries we need throughout the tutorial, including threading for concurrency, dataclasses for clean data modeling, and Rich for beautiful terminal output. We initialize the global OpenAI client, the Rich console, and set gpt-4o-mini as our default model for cost-efficient swarm execution.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class TaskStatus(str, Enum):\n   PENDING = \"pending\"\n   IN_PROGRESS = \"in_progress\"\n   COMPLETED = \"completed\"\n   BLOCKED = \"blocked\"\n   FAILED = \"failed\"\n\n\n\n\n@dataclass\nclass Task:\n   id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])\n   subject: str = \"\"\n   description: str = \"\"\n   owner: str = \"\"\n   status: TaskStatus = TaskStatus.PENDING\n   blocked_by: list = field(default_factory=list)\n   result: str = \"\"\n   created_at: str = field(default_factory=lambda: datetime.now().isoformat())\n   completed_at: Optional[str] = None\n\n\n   def to_dict(self):\n       return {\n           \"id\": self.id,\n           \"subject\": self.subject,\n           \"description\": self.description,\n           \"owner\": self.owner,\n           \"status\": self.status.value,\n           \"blocked_by\": self.blocked_by,\n           \"result\": self.result[:200] if self.result else \"\",\n       }\n\n\n\n\nclass TaskBoard:\n   def __init__(self):\n       self._tasks: dict[str, Task] = {}\n       self._lock = threading.Lock()\n\n\n   def create(self, subject: str, description: str = \"\", owner: str = \"\",\n              blocked_by: list = None) -> Task:\n       task = Task(\n           subject=subject,\n           description=description,\n           owner=owner,\n           blocked_by=blocked_by or [],\n       )\n       if task.blocked_by:\n           task.status = TaskStatus.BLOCKED\n       with self._lock:\n           self._tasks[task.id] = task\n       return task\n\n\n   def update_status(self, task_id: str, status: TaskStatus, result: str = \"\"):\n       with self._lock:\n           if task_id not in self._tasks:\n               return\n           task = self._tasks[task_id]\n           task.status = status\n           if result:\n               task.result = result\n           if status == TaskStatus.COMPLETED:\n               task.completed_at = datetime.now().isoformat()\n               self._resolve_dependencies(task_id)\n\n\n   def _resolve_dependencies(self, completed_id: str):\n       for t in self._tasks.values():\n           if completed_id in t.blocked_by:\n               t.blocked_by.remove(completed_id)\n               if not t.blocked_by and t.status == TaskStatus.BLOCKED:\n                   t.status = TaskStatus.PENDING\n\n\n   def get_tasks(self, owner: str = None, status: TaskStatus = None) -> list[Task]:\n       with self._lock:\n           tasks = list(self._tasks.values())\n       if owner:\n           tasks = [t for t in tasks if t.owner == owner]\n       if status:\n           tasks = [t for t in tasks if t.status == status]\n       return tasks\n\n\n   def get(self, task_id: str) -> Optional[Task]:\n       return self._tasks.get(task_id)\n\n\n   def all_completed(self) -> bool:\n       with self._lock:\n           return all(\n               t.status == TaskStatus.COMPLETED\n               for t in self._tasks.values()\n           )\n\n\n   def summary_json(self) -> str:\n       return json.dumps(\n           [t.to_dict() for t in self._tasks.values()],\n           indent=2,\n       )\n\n\n\n\n@dataclass\nclass Message:\n   sender: str\n   recipient: str\n   content: str\n   timestamp: str = field(default_factory=lambda: datetime.now().isoformat())\n\n\n\n\nclass InboxSystem:\n   def __init__(self):\n       self._inboxes: dict[str, list[Message]] = defaultdict(list)\n       self._lock = threading.Lock()\n\n\n   def send(self, sender: str, recipient: str, content: str):\n       msg = Message(sender=sender, recipient=recipient, content=content)\n       with self._lock:\n           self._inboxes[recipient].append(msg)\n\n\n   def broadcast(self, sender: str, content: str, exclude: list = None):\n       exclude = set(exclude or [])\n       exclude.add(sender)\n       with self._lock:\n           recipients = [r for r in self._inboxes.keys() if r not in exclude]\n       for r in recipients:\n           self.send(sender, r, content)\n\n\n   def receive(self, agent_name: str) -> list[Message]:\n       with self._lock:\n           msgs = self._inboxes.pop(agent_name, [])\n       return msgs\n\n\n   def peek(self, agent_name: str) -> list[Message]:\n       with self._lock:\n           return list(self._inboxes.get(agent_name, []))\n\n\n   def register(self, agent_name: str):\n       with self._lock:\n           if agent_name not in self._inboxes:\n               self._inboxes[agent_name] = []\n\n\n\n\n@dataclass\nclass AgentInfo:\n   name: str\n   role: str\n   status: str = \"active\"\n   tasks_completed: int = 0\n   spawned_at: str = field(default_factory=lambda: datetime.now().isoformat())\n\n\n\n\nclass TeamRegistry:\n   def __init__(self, team_name: str, description: str = \"\"):\n       self.team_name = team_name\n       self.description = description\n       self.agents: dict[str, AgentInfo] = {}\n       self._lock = threading.Lock()\n\n\n   def register(self, name: str, role: str):\n       with self._lock:\n           self.agents[name] = AgentInfo(name=name, role=role)\n\n\n   def update_status(self, name: str, status: str):\n       with self._lock:\n           if name in self.agents:\n               self.agents[name].status = status\n\n\n   def increment_completed(self, name: str):\n       with self._lock:\n           if name in self.agents:\n               self.agents[name].tasks_completed += 1<\/code><\/pre>\n<\/div>\n<\/div>\n
We build the three foundational systems that mirror ClawTeam\u2019s filesystem-based architecture: a TaskBoard with full status lifecycle and automatic dependency unblocking, an InboxSystem supporting point-to-point and broadcast messaging between agents, and a TeamRegistry that tracks every agent\u2019s role, status, and completed task count. We implement thread-safe locking across all three systems to allow multiple agents to read and write concurrently. We model tasks with a blocked_by dependency chain that automatically transitions blocked tasks to pending when their prerequisites complete, exactly how ClawTeam\u2019s CLI resolves task dependencies.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
SWARM_TOOLS = [\n   {\n       \"type\": \"function\",\n       \"function\": {\n           \"name\": \"task_update\",\n           \"description\": \"Update the status of a task. Use 'in_progress' when starting, 'completed' when done.\",\n           \"parameters\": {\n               \"type\": \"object\",\n               \"properties\": {\n                   \"task_id\": {\"type\": \"string\", \"description\": \"The task ID\"},\n                   \"status\": {\"type\": \"string\", \"enum\": [\"in_progress\", \"completed\", \"failed\"]},\n                   \"result\": {\"type\": \"string\", \"description\": \"Result or output of the task\"},\n               },\n               \"required\": [\"task_id\", \"status\"],\n           },\n       },\n   },\n   {\n       \"type\": \"function\",\n       \"function\": {\n           \"name\": \"inbox_send\",\n           \"description\": \"Send a message to another agent (e.g., 'leader' or a worker name).\",\n           \"parameters\": {\n               \"type\": \"object\",\n               \"properties\": {\n                   \"to\": {\"type\": \"string\", \"description\": \"Recipient agent name\"},\n                   \"message\": {\"type\": \"string\", \"description\": \"Message content\"},\n               },\n               \"required\": [\"to\", \"message\"],\n           },\n       },\n   },\n   {\n       \"type\": \"function\",\n       \"function\": {\n           \"name\": \"inbox_receive\",\n           \"description\": \"Check and consume all messages in your inbox.\",\n           \"parameters\": {\n               \"type\": \"object\",\n               \"properties\": {},\n           },\n       },\n   },\n   {\n       \"type\": \"function\",\n       \"function\": {\n           \"name\": \"task_list\",\n           \"description\": \"List tasks assigned to you or all team tasks.\",\n           \"parameters\": {\n               \"type\": \"object\",\n               \"properties\": {\n                   \"owner\": {\"type\": \"string\", \"description\": \"Filter by owner name (optional)\"},\n               },\n           },\n       },\n   },\n]\n\n\n\n\nclass SwarmAgent:\n\n\n   def __init__(\n       self,\n       name: str,\n       role: str,\n       system_prompt: str,\n       task_board: TaskBoard,\n       inbox: InboxSystem,\n       registry: TeamRegistry,\n   ):\n       self.name = name\n       self.role = role\n       self.system_prompt = system_prompt\n       self.task_board = task_board\n       self.inbox = inbox\n       self.registry = registry\n       self.conversation_history: list[dict] = []\n       self.inbox.register(name)\n       self.registry.register(name, role)\n\n\n   def _build_system_prompt(self) -> str:\n       coord_protocol = f\"\"\"\n## Coordination Protocol (auto-injected \u2014 you are agent '{self.name}')\n\n\nYou are part of an AI agent swarm. Your role: {self.role}\nYour name: {self.name}\n\n\nAvailable tools (equivalent to ClawTeam CLI):\n- task_list: Check your assigned tasks (like `clawteam task list`)\n- task_update: Update task status to in_progress\/completed\/failed (like `clawteam task update`)\n- inbox_send: Send messages to other agents (like `clawteam inbox send`)\n- inbox_receive: Check your inbox for messages (like `clawteam inbox receive`)\n\n\nWORKFLOW:\n1. Check your tasks with task_list\n2. Mark a task as in_progress when you start\n3. Do the work (think, analyze, produce output)\n4. Mark the task as completed with your result\n5. Send a summary message to 'leader' when done\n\"\"\"\n       return self.system_prompt + \"n\" + coord_protocol\n\n\n   def _handle_tool_call(self, tool_name: str, args: dict) -> str:\n       if tool_name == \"task_update\":\n           status = TaskStatus(args[\"status\"])\n           result = args.get(\"result\", \"\")\n           self.task_board.update_status(args[\"task_id\"], status, result)\n           if status == TaskStatus.COMPLETED:\n               self.registry.increment_completed(self.name)\n           return json.dumps({\"ok\": True, \"task_id\": args[\"task_id\"], \"new_status\": args[\"status\"]})\n\n\n       elif tool_name == \"inbox_send\":\n           self.inbox.send(self.name, args[\"to\"], args[\"message\"])\n           return json.dumps({\"ok\": True, \"sent_to\": args[\"to\"]})\n\n\n       elif tool_name == \"inbox_receive\":\n           msgs = self.inbox.receive(self.name)\n           if not msgs:\n               return json.dumps({\"messages\": [], \"note\": \"No new messages\"})\n           return json.dumps({\n               \"messages\": [\n                   {\"from\": m.sender, \"content\": m.content, \"time\": m.timestamp}\n                   for m in msgs\n               ]\n           })\n\n\n       elif tool_name == \"task_list\":\n           owner = args.get(\"owner\", self.name)\n           tasks = self.task_board.get_tasks(owner=owner)\n           return json.dumps({\"tasks\": [t.to_dict() for t in tasks]})\n\n\n       return json.dumps({\"error\": f\"Unknown tool: {tool_name}\"})\n\n\n   def run(self, user_message: str, max_iterations: int = 6) -> str:\n       self.conversation_history.append({\"role\": \"user\", \"content\": user_message})\n\n\n       for iteration in range(max_iterations):\n           try:\n               response = client.chat.completions.create(\n                   model=MODEL,\n                   messages=[\n                       {\"role\": \"system\", \"content\": self._build_system_prompt()},\n                       *self.conversation_history,\n                   ],\n                   tools=SWARM_TOOLS,\n                   tool_choice=\"auto\",\n                   temperature=0.4,\n               )\n           except Exception as e:\n               return f\"[API Error] {e}\"\n\n\n           choice = response.choices[0]\n           msg = choice.message\n\n\n           assistant_msg = {\"role\": \"assistant\", \"content\": msg.content or \"\"}\n           if msg.tool_calls:\n               assistant_msg[\"tool_calls\"] = [\n                   {\n                       \"id\": tc.id,\n                       \"type\": \"function\",\n                       \"function\": {\"name\": tc.function.name, \"arguments\": tc.function.arguments},\n                   }\n                   for tc in msg.tool_calls\n               ]\n           self.conversation_history.append(assistant_msg)\n\n\n           if not msg.tool_calls:\n               return msg.content or \"(No response)\"\n\n\n           for tc in msg.tool_calls:\n               fn_name = tc.function.name\n               fn_args = json.loads(tc.function.arguments)\n               result = self._handle_tool_call(fn_name, fn_args)\n               self.conversation_history.append({\n                   \"role\": \"tool\",\n                   \"tool_call_id\": tc.id,\n                   \"content\": result,\n               })\n\n\n       return \"(Agent reached max iterations)\"<\/code><\/pre>\n<\/div>\n<\/div>\n
We define four OpenAI function-calling tools, task_update, inbox_send, inbox_receive, and task_list, that serve as the agent\u2019s equivalent of ClawTeam\u2019s CLI commands. We construct the SwarmAgent class, which wraps an LLM reasoning loop that allows the agent to call tools multiple times before producing a final answer, mimicking how a real ClawTeam agent iteratively checks tasks, performs work, and reports results. We auto-inject a coordination protocol into every agent\u2019s system prompt, giving it awareness of its name, role, and the exact workflow it must follow, just as ClawTeam injects instructions into every spawned agent.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
LEADER_SYSTEM_PROMPT = \"\"\"You are the LEADER of an AI agent swarm (ClawTeam style).\n\n\nYour job:\n1. Receive a high-level goal from the human\n2. Decompose it into concrete sub-tasks with clear deliverables\n3. Assign tasks to specialized worker agents\n4. Monitor progress and synthesize the final result\n\n\nYou think strategically and delegate effectively. You don't do the detail work yourself \u2014\nyou coordinate the swarm.\n\n\nWhen producing the task plan, output valid JSON with this exact structure:\n{\n \"team_description\": \"Brief description of what this team does\",\n \"tasks\": [\n   {\n     \"subject\": \"Short task title\",\n     \"description\": \"Detailed instructions for the worker\",\n     \"worker_role\": \"The specialist role needed (e.g., 'Market Research Analyst')\",\n     \"worker_name\": \"Short snake_case name (e.g., 'market_analyst')\",\n     \"blocked_by_indices\": []\n   }\n ]\n}\n\n\nblocked_by_indices is a list of 0-based indices of other tasks that must complete first.\nCreate 3-5 tasks. Each task should be independently executable by a specialist.\n\"\"\"\n\n\n\n\ndef leader_plan_tasks(goal: str) -> dict:\n   console.print(Panel(\n       f\"[bold cyan]\"\ud83e\udd9e\" Leader Agent Activated[\/bold cyan]nn\"\n       f\"[white]Goal:[\/white] {goal}n\"\n       f\"[dim]Planning task decomposition...[\/dim]\",\n       border_style=\"cyan\",\n   ))\n\n\n   response = client.chat.completions.create(\n       model=MODEL,\n       messages=[\n           {\"role\": \"system\", \"content\": LEADER_SYSTEM_PROMPT},\n           {\"role\": \"user\", \"content\": f\"Decompose this goal into sub-tasks for a swarm of AI agents:nn{goal}\"},\n       ],\n       temperature=0.3,\n       response_format={\"type\": \"json_object\"},\n   )\n\n\n   plan_text = response.choices[0].message.content\n   plan = json.loads(plan_text)\n   return plan\n\n\n\n\ndef display_task_board(task_board: TaskBoard, registry: TeamRegistry):\n   table = Table(title=\"\"\ud83d\udccb\" Task Board (Kanban)\", show_lines=True, expand=True)\n   table.add_column(\"ID\", style=\"dim\", width=8)\n   table.add_column(\"Subject\", style=\"bold\")\n   table.add_column(\"Owner\", style=\"cyan\")\n   table.add_column(\"Status\", width=14)\n   table.add_column(\"Result Preview\", max_width=50)\n\n\n   status_colors = {\n       TaskStatus.PENDING: \"yellow\",\n       TaskStatus.IN_PROGRESS: \"blue\",\n       TaskStatus.COMPLETED: \"green\",\n       TaskStatus.BLOCKED: \"red\",\n       TaskStatus.FAILED: \"magenta\",\n   }\n\n\n   for t in task_board.get_tasks():\n       color = status_colors.get(t.status, \"white\")\n       status_str = f\"[{color}]\u25cf[\/{color}] {t.status.value}\"\n       result_preview = (t.result[:47] + \"...\") if len(t.result) > 50 else t.result\n       table.add_row(t.id, t.subject, t.owner, status_str, result_preview)\n\n\n   console.print(table)\n\n\n   agent_table = Table(title=\"\"\ud83e\udd16\" Agent Roster\", show_lines=True)\n   agent_table.add_column(\"Agent\", style=\"bold cyan\")\n   agent_table.add_column(\"Role\")\n   agent_table.add_column(\"Status\")\n   agent_table.add_column(\"Tasks Done\", justify=\"center\")\n\n\n   for a in registry.agents.values():\n       status_emoji = {\"active\": \"\"\ud83d\udd04\"\", \"idle\": \"\"\ud83d\ude34\"\", \"completed\": \"\"\u2705\"\", \"failed\": \"\"\u274c\"\"}.get(a.status, \"\"\u2753\"\")\n       agent_table.add_row(a.name, a.role, f\"{status_emoji} {a.status}\", str(a.tasks_completed))\n\n\n   console.print(agent_table)\n\n\n\n\ndef run_swarm(goal: str):\n   console.print(Panel(\n       \"[bold white on blue] \"\ud83e\udd9e\" ClawTeam Swarm Intelligence \u2014 Tutorial Run [\/bold white on blue]\",\n       expand=True,\n   ))\n   console.print(f\"n[bold]Human Goal:[\/bold] {goal}n\")\n\n\n   console.rule(\"[bold cyan]Phase 1: Leader Decomposes Goal into Tasks\")\n   plan = leader_plan_tasks(goal)\n\n\n   console.print(f\"n[bold green]Team:[\/bold green] {plan.get('team_description', 'N\/A')}\")\n   console.print(f\"[bold green]Tasks planned:[\/bold green] {len(plan['tasks'])}n\")\n\n\n   console.rule(\"[bold cyan]Phase 2: Spawning Swarm Infrastructure\")\n\n\n   task_board = TaskBoard()\n   inbox = InboxSystem()\n   registry = TeamRegistry(team_name=\"swarm-alpha\", description=plan.get(\"team_description\", \"\"))\n   registry.register(\"leader\", \"Swarm Leader \/ Orchestrator\")\n   inbox.register(\"leader\")\n\n\n   task_map = {}\n   for i, t in enumerate(plan[\"tasks\"]):\n       blocked_by_ids = [task_map[idx] for idx in t.get(\"blocked_by_indices\", []) if idx in task_map]\n       task = task_board.create(\n           subject=t[\"subject\"],\n           description=t[\"description\"],\n           owner=t[\"worker_name\"],\n           blocked_by=blocked_by_ids,\n       )\n       task_map[i] = task.id\n       console.print(\n           f\"  [cyan]\"\ud83d\udccb\" Task {task.id}[\/cyan] \u2192 [bold]{t['subject']}[\/bold] \"\n           f\"(assigned to [yellow]{t['worker_name']}[\/yellow])\"\n           + (f\" [red]blocked by {blocked_by_ids}[\/red]\" if blocked_by_ids else \"\")\n       )\n\n\n   console.rule(\"[bold cyan]Phase 3: Spawning Worker Agents\")\n   agents: dict[str, SwarmAgent] = {}\n\n\n   for t in plan[\"tasks\"]:\n       name = t[\"worker_name\"]\n       if name not in agents:\n           worker_prompt = f\"\"\"You are a specialist AI agent.\nYour role: {t['worker_role']}\nYour name: {name}\n\n\nYou will receive tasks and must:\n1. Check your task list\n2. Mark each task as in_progress\n3. Do thorough, high-quality work on the task\n4. Report your detailed findings\/output as the task result\n5. Mark the task as completed\n6. Message the leader with a summary\n\n\nBe thorough, specific, and actionable in your outputs.\"\"\"\n\n\n           agent = SwarmAgent(\n               name=name,\n               role=t[\"worker_role\"],\n               system_prompt=worker_prompt,\n               task_board=task_board,\n               inbox=inbox,\n               registry=registry,\n           )\n           agents[name] = agent\n           console.print(\n               f\"  [green]\"\ud83e\udd16\" Spawned:[\/green] [bold]{name}[\/bold] \u2014 {t['worker_role']}\"\n           )\n\n\n   console.print(f\"n[bold]Total agents spawned:[\/bold] {len(agents)} workers + 1 leadern\")\n\n\n   console.rule(\"[bold cyan]Phase 4: Swarm Execution\")\n\n\n   max_rounds = 3\n   for round_num in range(max_rounds):\n       console.print(f\"n[bold magenta]\u2500\u2500 Round {round_num + 1} \u2500\u2500[\/bold magenta]\")\n\n\n       active_this_round = False\n       for agent_name, agent in agents.items():\n           my_tasks = task_board.get_tasks(owner=agent_name)\n           pending = [t for t in my_tasks if t.status in (TaskStatus.PENDING, TaskStatus.IN_PROGRESS)]\n\n\n           if not pending:\n               continue\n\n\n           active_this_round = True\n           for task in pending:\n               if task.status == TaskStatus.BLOCKED:\n                   continue\n\n\n               console.print(\n                   f\"n  [cyan]\"\u25b6\" Agent [bold]{agent_name}[\/bold] working on:[\/cyan] \"\n                   f\"{task.subject} (ID: {task.id})\"\n               )\n\n\n               prompt = (\n                   f\"You have a task to complete.n\"\n                   f\"Task ID: {task.id}n\"\n                   f\"Subject: {task.subject}n\"\n                   f\"Description: {task.description}nn\"\n                   f\"Steps:n\"\n                   f\"1. Call task_update to mark task '{task.id}' as 'in_progress'n\"\n                   f\"2. Do the work and produce a detailed, high-quality resultn\"\n                   f\"3. Call task_update to mark task '{task.id}' as 'completed' with your resultn\"\n                   f\"4. Call inbox_send to message 'leader' with a brief summaryn\"\n               )\n\n\n               result = agent.run(prompt)\n               console.print(f\"  [green]\"\u2705\" Agent {agent_name} finished.[\/green]\")\n\n\n               task_obj = task_board.get(task.id)\n               if task_obj and task_obj.status == TaskStatus.COMPLETED:\n                   registry.update_status(agent_name, \"completed\")\n                   console.print(f\"  [dim]Result preview: {task_obj.result[:100]}...[\/dim]\")\n\n\n       if not active_this_round or task_board.all_completed():\n           break\n\n\n   console.rule(\"[bold cyan]Phase 5: Leader Synthesizes Final Report\")\n\n\n   all_results = []\n   for t in task_board.get_tasks():\n       all_results.append({\n           \"task\": t.subject,\n           \"worker\": t.owner,\n           \"status\": t.status.value,\n           \"result\": t.result,\n       })\n\n\n   leader_msgs = inbox.receive(\"leader\")\n   msg_summary = \"n\".join(\n       [f\"From {m.sender}: {m.content}\" for m in leader_msgs]\n   )\n\n\n   synthesis_prompt = f\"\"\"You are the swarm leader. Your workers have completed their tasks.\nHere are all the results:\n\n\n{json.dumps(all_results, indent=2)}\n\n\nMessages from workers:\n{msg_summary}\n\n\nSynthesize all worker outputs into a coherent, comprehensive final report.\nThe report should:\n1. Summarize the overall findings\n2. Highlight key insights from each worker\n3. Identify connections between different workers' findings\n4. Provide actionable recommendations\n5. Note any gaps or areas for further investigation\n\n\nFormat it as a well-structured report with clear sections.\"\"\"\n\n\n   console.print(\"[dim]Leader synthesizing final report...[\/dim]n\")\n\n\n   synthesis_response = client.chat.completions.create(\n       model=MODEL,\n       messages=[\n           {\"role\": \"system\", \"content\": \"You are a senior analyst synthesizing a team's research into a final report.\"},\n           {\"role\": \"user\", \"content\": synthesis_prompt},\n       ],\n       temperature=0.3,\n   )\n\n\n   final_report = synthesis_response.choices[0].message.content\n\n\n   console.rule(\"[bold cyan]Phase 6: Final Dashboard\")\n   display_task_board(task_board, registry)\n\n\n   console.print()\n   console.rule(\"[bold green]\"\ud83e\udd9e\" SWARM FINAL REPORT\")\n   console.print(Panel(\n       final_report,\n       title=\"[bold]Synthesized Report[\/bold]\",\n       border_style=\"green\",\n       expand=True,\n   ))\n\n\n   completed = len([t for t in task_board.get_tasks() if t.status == TaskStatus.COMPLETED])\n   total = len(task_board.get_tasks())\n   console.print(f\"n[bold]\"\ud83d\udcca\" Swarm Stats:[\/bold]\")\n   console.print(f\"   Tasks completed: {completed}\/{total}\")\n   console.print(f\"   Agents used: {len(agents)}\")\n   console.print(f\"   Execution rounds: {round_num + 1}\")\n   console.print(f\"   Model: {MODEL}\")\n   console.print()\n\n\n   return final_report<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement the leader agent, which takes a high-level human goal and produces structured JSON output that decomposes it into 3\u20135 concrete subtasks, with worker role assignments and dependency chains. We build the full run_swarm pipeline that executes six phases: leader planning, infrastructure initialization, worker spawning, multi-round task execution with dependency resolution, leader synthesis of all results into a final report, and a Rich-powered dashboard display. We render a live kanban board and agent roster at the end, giving visibility into every task\u2019s status, every agent\u2019s contribution, and the synthesized final output, replicating ClawTeam\u2019s board show monitoring experience.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
TEMPLATES = {\n   \"hedge_fund\": {\n       \"name\": \"AI Hedge Fund\",\n       \"description\": \"Multi-analyst investment research team (inspired by ClawTeam's hedge fund template)\",\n       \"goal_template\": \"Analyze {stock} for investment potential. Cover: value investing fundamentals, \"\n                        \"growth metrics & disruption potential, technical chart indicators, \"\n                        \"financial ratio analysis, and sentiment\/news analysis. \"\n                        \"Produce a final buy\/hold\/sell recommendation with confidence level.\",\n   },\n   \"research\": {\n       \"name\": \"Research Swarm\",\n       \"description\": \"Multi-perspective deep research team\",\n       \"goal_template\": \"Conduct comprehensive research on: {topic}. Cover: background & current state, \"\n                        \"key challenges & open problems, recent breakthroughs & innovations, \"\n                        \"future outlook & predictions, and practical applications.\",\n   },\n   \"engineering\": {\n       \"name\": \"Engineering Team\",\n       \"description\": \"Software architecture & design team\",\n       \"goal_template\": \"Design the architecture for: {project}. Cover: requirements analysis, \"\n                        \"system architecture & component design, API design & data models, \"\n                        \"security considerations, and deployment & scaling strategy.\",\n   },\n}\n\n\n\n\ndef list_templates():\n   table = Table(title=\"\"\ud83c\udfaa\" Available Team Templates\", show_lines=True)\n   table.add_column(\"Key\", style=\"bold cyan\")\n   table.add_column(\"Name\", style=\"bold\")\n   table.add_column(\"Description\")\n\n\n   for key, tmpl in TEMPLATES.items():\n       table.add_row(key, tmpl[\"name\"], tmpl[\"description\"])\n   console.print(table)\n\n\n\n\ndef launch_template(template_key: str, **kwargs) -> str:\n   tmpl = TEMPLATES.get(template_key)\n   if not tmpl:\n       console.print(f\"[red]Template '{template_key}' not found.[\/red]\")\n       list_templates()\n       return \"\"\n\n\n   goal = tmpl[\"goal_template\"].format(**kwargs)\n   console.print(f\"n[bold]\"\ud83c\udfaa\" Launching template:[\/bold] {tmpl['name']}\")\n   console.print(f\"[bold]Goal:[\/bold] {goal}n\")\n   return run_swarm(goal)\n\n\n\n\ndef interactive_demo():\n   console.print(Panel(\n       \"[bold white]\"\ud83e\udd9e\" ClawTeam Agent Swarm Intelligence \u2014 Interactive Tutorial[\/bold white]nn\"\n       \"This tutorial implements the core concepts from ClawTeam:n\"\n       \"  \u2022 Leader agent decomposes goals into tasksn\"\n       \"  \u2022 Worker agents are spawned with specialized rolesn\"\n       \"  \u2022 Agents communicate via inbox messagingn\"\n       \"  \u2022 Task dependencies auto-resolven\"\n       \"  \u2022 A dashboard monitors swarm progressn\"\n       \"  \u2022 The leader synthesizes a final reportnn\"\n       \"[dim]Powered by OpenAI + ClawTeam architecture patterns[\/dim]\",\n       border_style=\"cyan\",\n       title=\"Welcome\",\n   ))\n\n\n   console.print(\"n[bold]Choose a demo:[\/bold]n\")\n   console.print(\"  [cyan]1.[\/cyan] \"\ud83d\udcb0\" AI Hedge Fund \u2014 Analyze a stock with 5 specialist agents\")\n   console.print(\"  [cyan]2.[\/cyan] \"\ud83d\udd2c\" Research Swarm \u2014 Deep-dive into any topic\")\n   console.print(\"  [cyan]3.[\/cyan] \"\ud83c\udfd7\"  Engineering Team \u2014 Design a software system\")\n   console.print(\"  [cyan]4.[\/cyan] \"\ud83c\udfaf\" Custom Goal \u2014 Enter your own goal for the swarm\")\n   console.print(\"  [cyan]5.[\/cyan] \"\ud83d\udccb\" List Templates\")\n   console.print()\n\n\n   choice = input(\"Enter your choice (1-5): \").strip()\n\n\n   if choice == \"1\":\n       stock = input(\"Enter stock ticker(s) to analyze (e.g., NVDA): \").strip() or \"NVDA\"\n       launch_template(\"hedge_fund\", stock=stock)\n\n\n   elif choice == \"2\":\n       topic = input(\"Enter research topic: \").strip() or \"The future of AI agents and multi-agent systems\"\n       launch_template(\"research\", topic=topic)\n\n\n   elif choice == \"3\":\n       project = input(\"Enter project to design: \").strip() or \"A real-time collaborative document editor\"\n       launch_template(\"engineering\", project=project)\n\n\n   elif choice == \"4\":\n       goal = input(\"Enter your goal for the swarm: \").strip()\n       if goal:\n           run_swarm(goal)\n       else:\n           console.print(\"[red]No goal entered.[\/red]\")\n\n\n   elif choice == \"5\":\n       list_templates()\n\n\n   else:\n       console.print(\"[red]Invalid choice. Running default hedge fund demo...[\/red]\")\n       launch_template(\"hedge_fund\", stock=\"NVDA\")\n\n\n\n\nif __name__ == \"__main__\":\n   interactive_demo()<\/code><\/pre>\n<\/div>\n<\/div>\n
We create three pre-built team templates, AI Hedge Fund, Research Swarm, and Engineering Team, that mirror ClawTeam\u2019s TOML-based team archetypes and let users launch a full multi-agent workflow with a single command. We build an interactive menu that presents all five options (three templates, custom goal entry, and template listing) and handles user input gracefully with sensible defaults. We wire everything to the __main__ entry point so that running the final cell kicks off the entire interactive experience, from choosing a demo to watching agents coordinate and deliver a synthesized report.<\/p>\n
In conclusion, we demonstrated through this tutorial that the core principles behind ClawTeam\u2019s Agent Swarm Intelligence, task decomposition, agent spawning, dependency-aware scheduling, inter-agent messaging, and leader-driven synthesis, can be faithfully reproduced using OpenAI\u2019s function-calling API and Python. We show that even without ClawTeam\u2019s native infrastructure of tmux sessions, git worktrees, and filesystem-based message queues, the architectural patterns themselves are powerful enough to enable meaningful multi-agent collaboration, with specialized workers coordinating autonomously under a leader\u2019s orchestration.<\/p>\n
\n
Check out\u00a0Full Notebook here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0120k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post A Coding Implementation Showcasing ClawTeam\u2019s Multi-Agent Swarm Orchestration with OpenAI Function Calling<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this comprehensive tutorial…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-588","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/588","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=588"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/588\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=588"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=588"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=588"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}