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"]},
"outcome": {"kind": "string", "description": "End result or output of the duty"},
},
"required": ["task_id", "status"],
},
},
},
{
"kind": "perform",
"perform": {
"identify": "inbox_send",
"description": "Ship a message to a different agent (e.g., 'chief' or a employee identify).",
"parameters": {
"kind": "object",
"properties": {
"to": {"kind": "string", "description": "Recipient agent identify"},
"message": {"kind": "string", "description": "Message content material"},
},
"required": ["to", "message"],
},
},
},
{
"kind": "perform",
"perform": {
"identify": "inbox_receive",
"description": "Verify and devour all messages in your inbox.",
"parameters": {
"kind": "object",
"properties": {},
},
},
},
{
"kind": "perform",
"perform": {
"identify": "task_list",
"description": "Checklist duties assigned to you or all staff duties.",
"parameters": {
"kind": "object",
"properties": {
"proprietor": {"kind": "string", "description": "Filter by proprietor identify (optionally available)"},
},
},
},
},
]
class SwarmAgent:
def __init__(
self,
identify: str,
function: str,
system_prompt: str,
task_board: TaskBoard,
inbox: InboxSystem,
registry: TeamRegistry,
):
self.identify = identify
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(identify)
self.registry.register(identify, function)
def _build_system_prompt(self) -> str:
coord_protocol = f"""
## Coordination Protocol (auto-injected — you're agent '{self.identify}')
You might be a part of an AI agent swarm. Your function: {self.function}
Your identify: {self.identify}
Out there instruments (equal to ClawTeam CLI):
- task_list: Verify your assigned duties (like `clawteam job listing`)
- task_update: Replace job standing to in_progress/accomplished/failed (like `clawteam job replace`)
- inbox_send: Ship messages to different brokers (like `clawteam inbox ship`)
- inbox_receive: Verify your inbox for messages (like `clawteam inbox obtain`)
WORKFLOW:
1. Verify your duties with task_list
2. Mark a job as in_progress while you begin
3. Do the work (suppose, analyze, produce output)
4. Mark the duty as accomplished together with your outcome
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"])
outcome = args.get("outcome", "")
self.task_board.update_status(args["task_id"], standing, outcome)
if standing == TaskStatus.COMPLETED:
self.registry.increment_completed(self.identify)
return json.dumps({"okay": True, "task_id": args["task_id"], "new_status": args["status"]})
elif tool_name == "inbox_send":
self.inbox.ship(self.identify, args["to"], args["message"])
return json.dumps({"okay": True, "sent_to": args["to"]})
elif tool_name == "inbox_receive":
msgs = self.inbox.obtain(self.identify)
if not msgs:
return json.dumps({"messages": [], "be aware": "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("proprietor", self.identify)
duties = self.task_board.get_tasks(proprietor=proprietor)
return json.dumps({"duties": [t.to_dict() for t in tasks]})
return json.dumps({"error": f"Unknown instrument: {tool_name}"})
def run(self, user_message: str, max_iterations: int = 6) -> str:
self.conversation_history.append({"function": "consumer", "content material": user_message})
for iteration in vary(max_iterations):
attempt:
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}"
selection = response.decisions[0]
msg = selection.message
assistant_msg = {"function": "assistant", "content material": 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.perform.identify
fn_args = json.hundreds(tc.perform.arguments)
outcome = self._handle_tool_call(fn_name, fn_args)
self.conversation_history.append({
"function": "instrument",
"tool_call_id": tc.id,
"content material": outcome,
})
return "(Agent reached max iterations)"
