agent_broadcast

Instructions

Implementation Reference

• python/src/agentlink_mcp/server.py:2085-2140 (handler)

  The main MCP tool handler for 'agent_broadcast'. This async function accepts a BroadcastInput params, creates a broadcast payload, sends it via WebSocket to the Cloudflare relay, handles ACK responses, and manages retry queue for offline scenarios.

  @mcp.tool(name="agent_broadcast")
  async def agent_broadcast(params: BroadcastInput) -> str:
      """
      Kirim pesan ke semua agent di room via Cloudflare relay.
  
      Args:
          params: message, msg_type
      Returns:
          str: JSON status broadcast
      """
      payload = {"type": "broadcast",
          "content": params.message, "msg_type": params.msg_type}
      if current_room is None:
          return json.dumps({"success": False, "error": "Tidak terhubung ke room. Jalankan room_join dulu."})
      if ws_conn is None:
          queued = _enqueue_retry_action("broadcast", payload, reason="WebSocket tidak terhubung")
          return json.dumps({"success": False, "queued_for_retry": True,
              "retry_queue_id": queued["retry_id"],
              "retry_queue_count": len(_retry_queue()),
              "message": "Broadcast disimpan di retry queue lokal hingga koneksi pulih."})
      try:
          request_id, ack = await _await_ack(payload)
          if ack is None:
              queued = _enqueue_retry_action("broadcast", payload, reason="ACK timeout")
              return json.dumps({"success": False, "request_id": request_id,
                  "queued_for_retry": True,
                  "retry_queue_id": queued["retry_id"],
                  "retry_queue_count": len(_retry_queue()),
                  "message": "ACK timeout. Broadcast disimpan di retry queue lokal."})
          if ack.get("delivered", False):
              return json.dumps({"success": ack.get("accepted", False),
                  "accepted": ack.get("accepted", False),
                  "delivered": ack.get("delivered", False),
                  "recipient_count": ack.get("recipient_count", 0),
                  "request_id": request_id,
                  "message_id": ack.get("message_id"),
                  "sequence": ack.get("sequence"),
                  "message": "Broadcast diterima relay."})
          queued = _enqueue_retry_action("broadcast", payload, reason="Belum ada penerima aktif")
          return json.dumps({"success": False,
              "accepted": ack.get("accepted", False),
              "delivered": ack.get("delivered", False),
              "recipient_count": ack.get("recipient_count", 0),
              "request_id": request_id,
              "message_id": ack.get("message_id"),
              "sequence": ack.get("sequence"),
              "queued_for_retry": True,
              "retry_queue_id": queued["retry_id"],
              "retry_queue_count": len(_retry_queue()),
              "message": "Broadcast disimpan di retry queue lokal sampai ada penerima aktif."})
      except Exception as e:
          queued = _enqueue_retry_action("broadcast", payload, reason=str(e))
          return json.dumps({"success": False, "error": str(e),
              "queued_for_retry": True,
              "retry_queue_id": queued["retry_id"],
              "retry_queue_count": len(_retry_queue())})

• python/src/agentlink_mcp/server.py:1314-1317 (schema)

  BroadcastInput Pydantic model defining the input schema for agent_broadcast tool. Requires 'message' (min 1 char) and optional 'msg_type' (default 'text').

  class BroadcastInput(BaseModel):
      model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
      message:  str = Field(..., description="Isi pesan ke semua peer", min_length=1)
      msg_type: str = Field(default="text", description="Tipe pesan")

• src/index.ts:535-563 (handler)

  Server-side WebSocket handler for 'broadcast' message type in the AgentLinkRoom Durable Object. Creates a room message with broadcast=true, sends to all connected agents except sender, and returns an ACK with recipient count.

  if (msg.type === "broadcast") {
    const timestamp = new Date().toISOString();
    const sequence = await this.nextSequence();
    const payload = createRoomMessage({
      roomId,
      sequence,
      timestamp,
      from: agentId,
      fromName: name,
      content: msg.content,
      msgType: typeof msg.msg_type === "string" ? msg.msg_type : "text",
      broadcast: true,
    });
    const recipientCount = this.broadcast(ws, JSON.stringify(payload));
    const requestId = typeof msg.request_id === "string" ? msg.request_id : undefined;
  
    ws.send(JSON.stringify(createAck({
      action: "broadcast",
      roomId,
      requestId,
      delivered: recipientCount > 0,
      recipientCount,
      timestamp,
      messageId: payload.message_id,
      sequence: payload.sequence,
      broadcast: true,
    })));
    return;
  }

• src/message-protocol.ts:60-82 (helper)

  createRoomMessage helper function used by the broadcast handler to construct the message payload with room_id, sequence, timestamp, from, content, msg_type, and broadcast flag.

  export function createRoomMessage(params: {
    roomId: string;
    sequence: number;
    timestamp: string;
    from: string;
    fromName: string;
    content: unknown;
    msgType?: string;
    broadcast?: boolean;
  }): RoomMessagePayload {
    return {
      type: "message",
      room_id: params.roomId,
      message_id: buildMessageId(params.roomId, params.sequence),
      sequence: params.sequence,
      timestamp: params.timestamp,
      from: params.from,
      from_name: params.fromName,
      content: params.content,
      msg_type: params.msgType || "text",
      broadcast: params.broadcast === true,
    };
  }

• src/message-protocol.ts:8-15 (schema)

  RoomMessagePayload TypeScript interface defining the structure of broadcast messages including type, from, from_name, content, msg_type, and broadcast boolean flag.

  export interface RoomMessagePayload extends SequencedRoomPayload {
    type: "message";
    from: string;
    from_name: string;
    content: unknown;
    msg_type: string;
    broadcast: boolean;
  }