ClawTeam’s Multi-Agent Flow Orchestration is Demonstrated with OpenAI function calling in this Coding Implementation

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,
       name: str,
       role: str,
       system_prompt: str,
       task_board: TaskBoard,
       inbox: InboxSystem,
       registry: TeamRegistry,
   ):
 Self-name = Name
 Self-role = Role
       self.system_prompt = system_prompt
       self.task_board = task_board
 Self.inbox is the same as inbox
 Self-registry = Registry
       self.conversation_history: list[dict] = []
       self.inbox.register(name)
       self.registry.register(name, role)


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


You're part of a swarm of AI agents. Your role: {self.role}
Your name: {self.name}


Tools available for ClawTeam CLI:
Check the task_list to see what you have been assigned (e.g. 'clawteam task-list).
- task_update: Update task status to in_progress/completed/failed (like `clawteam task update`)
- Inbox_send send messages (like "clawteam mail to inbox")
Inbox_receive : check your mailbox for any messages, such as clawteam receive inbox


WORKFLOW:
1. Use task_list to check your list of tasks
2. Assign a status of in_progress to a job when you first start
3. Work (think, analyse, and produce output).
4. The task is complete when you mark it as such.
5. Once done, you can send an email to the leader.
"""
       return self.system_prompt + "n" + coord_protocol


   def _handle_tool_call(self, tool_name: str, args: dict) -> str:
 if tool_name== "task_update":
 Status = TaskStatus (args)["status"])
 If result is args.get(), then the argument has been retrieved."result", "")
           self.task_board.update_status(args["task_id"], status, result)
           if status == TaskStatus.COMPLETED:
               self.registry.increment_completed(self.name)
           return json.dumps({"ok": True, "task_id"The args["task_id"], "new_status"The. args["status"]})


 If tool_name is equal to "" then the following will be displayed: "inbox_send":
           self.inbox.send(self.nameThe args["to"], args["message"])
           return json.dumps({"ok": True, "sent_to"Args["to"]})


 If tool_name is equal to "" then the following will be displayed: "inbox_receive":
           msgs = self.inbox.receive(self.name)
 If not messages:
               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
               ]
           })


 If tool_name is equal to "" then the following will be displayed: "task_list":
 Owner = args.get()"owner", self.name)
           tasks = self.task_board.get_tasks(owner=owner)
           return json.dumps({"tasks": [t.to_dict() for t in tasks]})


       return json.dumps({"error"""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 range(max_iterations):
           try:
               response = client.chat.completions.create(
                   model=MODEL,
                   messages=[
                       {"role": "system", "content": self._build_system_prompt()},
                       *self.conversation_history,
                   ],
                   tools=SWARM_TOOLS,
                   tool_choice="auto",
                   temperature=0.4,
               )
 "except Exception" is a phrase that means:
 Return f"[API Error] {e}"


 Choice = Response.[0]
 Choose.message


           assistant_msg = {"role": "assistant", "content"Or msg.content ""}
 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 message.content "(No response)"


 For tc calls in msg.tool_calls
               fn_name = tc.function.name
               fn_args = json.loads(tc.function.arguments)
               result = self._handle_tool_call(fn_name, fn_args)
               self.conversation_history.append({
                   "role": "tool",
                   "tool_call_id": tc.id,
                   "content": result,
               })


 You can return to your original language by clicking here. "(Agent reached max iterations)"