A Coding Implementation Showcasing ClawTeam’s Multi-Agent Swarm Orchestration with OpenAI Operate Calling

SWARM_TOOLS = [
   {
       "type": "function",
       "function": {
           "name": "task_update",
           "description": "Update the status of a task. Use 'in_progress' when starting, 'completed' when done.",
           "parameters": {
               "type": "object",
               "properties": {
                   "task_id": {"type": "string", "description": "The task ID"},
                   "status": {"type": "string", "enum": ["in_progress", "completed", "failed"]},
                   "result": {"type": "string", "description": "Result or output of the task"},
               },
               "required": ["task_id", "status"],
           },
       },
   },
   {
       "type": "function",
       "function": {
           "name": "inbox_send",
           "description": "Send a message to another agent (e.g., 'leader' or a worker name).",
           "parameters": {
               "type": "object",
               "properties": {
                   "to": {"type": "string", "description": "Recipient agent name"},
                   "message": {"type": "string", "description": "Message content"},
               },
               "required": ["to", "message"],
           },
       },
   },
   {
       "type": "function",
       "function": {
           "name": "inbox_receive",
           "description": "Check and consume all messages in your inbox.",
           "parameters": {
               "type": "object",
               "properties": {},
           },
       },
   },
   {
       "type": "function",
       "function": {
           "name": "task_list",
           "description": "List tasks assigned to you or all team tasks.",
           "parameters": {
               "type": "object",
               "properties": {
                   "owner": {"type": "string", "description": "Filter by owner name (optional)"},
               },
           },
       },
   },
]




class SwarmAgent:


   def __init__(
       self,
       title: str,
       function: str,
       system_prompt: str,
       task_board: TaskBoard,
       inbox: InboxSystem,
       registry: TeamRegistry,
   ):
       self.title = title
       self.function = function
       self.system_prompt = system_prompt
       self.task_board = task_board
       self.inbox = inbox
       self.registry = registry
       self.conversation_history: listing[dict] = []
       self.inbox.register(title)
       self.registry.register(title, function)


   def _build_system_prompt(self) -> str:
       coord_protocol = f"""
## Coordination Protocol (auto-injected — you might be agent '{self.title}')


You might be a part of an AI agent swarm. Your function: {self.function}
Your title: {self.title}


Accessible instruments (equal to ClawTeam CLI):
- task_list: Test your assigned duties (like `clawteam process listing`)
- task_update: Replace process standing to in_progress/accomplished/failed (like `clawteam process replace`)
- inbox_send: Ship messages to different brokers (like `clawteam inbox ship`)
- inbox_receive: Test your inbox for messages (like `clawteam inbox obtain`)


WORKFLOW:
1. Test your duties with task_list
2. Mark a process as in_progress once you begin
3. Do the work (suppose, analyze, produce output)
4. Mark the duty as accomplished together with your consequence
5. Ship a abstract message to 'chief' when performed
"""
       return self.system_prompt + "n" + coord_protocol


   def _handle_tool_call(self, tool_name: str, args: dict) -> str:
       if tool_name == "task_update":
           standing = TaskStatus(args["status"])
           consequence = args.get("result", "")
           self.task_board.update_status(args["task_id"], standing, consequence)
           if standing == TaskStatus.COMPLETED:
               self.registry.increment_completed(self.title)
           return json.dumps({"ok": True, "task_id": args["task_id"], "new_status": args["status"]})


       elif tool_name == "inbox_send":
           self.inbox.ship(self.title, args["to"], args["message"])
           return json.dumps({"ok": True, "sent_to": args["to"]})


       elif tool_name == "inbox_receive":
           msgs = self.inbox.obtain(self.title)
           if not msgs:
               return json.dumps({"messages": [], "note": "No new messages"})
           return json.dumps({
               "messages": [
                   {"from": m.sender, "content": m.content, "time": m.timestamp}
                   for m in msgs
               ]
           })


       elif tool_name == "task_list":
           proprietor = args.get("owner", self.title)
           duties = self.task_board.get_tasks(proprietor=proprietor)
           return json.dumps({"tasks": [t.to_dict() for t in tasks]})


       return json.dumps({"error": f"Unknown tool: {tool_name}"})


   def run(self, user_message: str, max_iterations: int = 6) -> str:
       self.conversation_history.append({"role": "user", "content": user_message})


       for iteration in vary(max_iterations):
           strive:
               response = shopper.chat.completions.create(
                   mannequin=MODEL,
                   messages=[
                       {"role": "system", "content": self._build_system_prompt()},
                       *self.conversation_history,
                   ],
                   instruments=SWARM_TOOLS,
                   tool_choice="auto",
                   temperature=0.4,
               )
           besides Exception as e:
               return f"[API Error] {e}"


           alternative = response.selections[0]
           msg = alternative.message


           assistant_msg = {"role": "assistant", "content": msg.content material or ""}
           if msg.tool_calls:
               assistant_msg["tool_calls"] = [
                   {
                       "id": tc.id,
                       "type": "function",
                       "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                   }
                   for tc in msg.tool_calls
               ]
           self.conversation_history.append(assistant_msg)


           if not msg.tool_calls:
               return msg.content material or "(No response)"


           for tc in msg.tool_calls:
               fn_name = tc.operate.title
               fn_args = json.masses(tc.operate.arguments)
               consequence = self._handle_tool_call(fn_name, fn_args)
               self.conversation_history.append({
                   "role": "tool",
                   "tool_call_id": tc.id,
                   "content": consequence,
               })


       return "(Agent reached max iterations)"