Camel.toolkits.message agent toolkit

​

AgentMessage

class AgentMessage(BaseMessage):

​

init

def __init__(
    self,
    message_id: str,
    sender_id: str,
    receiver_id: str,
    content: str,
    timestamp: float,
    reply_to: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
):

​

AgentCommunicationToolkit

class AgentCommunicationToolkit(BaseToolkit):

• agents (Optional[Dict[str, ChatAgent]]): Dictionary mapping agent IDs to ChatAgent instances. (default: :obj:None)
• timeout (Optional[float]): Maximum execution time for operations in seconds. (default: :obj:None)
• max_message_history (int): Maximum messages to keep per agent. (default: :obj:100)
• get_response (bool): Whether to get responses from receiving agents by default. (default: :obj:False)

​

init

def __init__(
    self,
    agents: Optional[Dict[str, 'ChatAgent']] = None,
    timeout: Optional[float] = None,
    max_message_history: int = 100,
    get_response: bool = False
):

​

register_agent

def register_agent(self, agent_id: str, agent: 'ChatAgent'):

• agent_id (str): Unique identifier for the agent.
• agent (ChatAgent): The ChatAgent instance to register.

​

_find_agent_id

def _find_agent_id(self, agent_id: str):

​

send_message

def send_message(
    self,
    message: str,
    receiver_id: str,
    sender_id: str = 'system',
    reply_to: Optional[str] = None,
    metadata_json: Optional[str] = None
):

• message (str): The content of the message to send.
• receiver_id (str): The unique identifier of the agent to receive the message. Use list_available_agents() to find the ID of the agent you want to talk to.
• sender_id (str): The unique identifier of the agent sending the message. This is typically your agent’s ID. (default: :obj:"system")
• reply_to (Optional[str]): The ID of a previous message this new message is a reply to. This helps create conversation threads. (default: :obj:None)
• metadata_json (Optional[str]): A JSON string containing extra information about the message. (default: :obj:None)

​

_deliver_message

def _deliver_message(self, message: AgentMessage, get_response: bool):

​

broadcast_message

def broadcast_message(
    self,
    message: str,
    sender_id: str = 'system',
    exclude_agents: Optional[List[str]] = None
):

• message (str): The content of the message to broadcast.
• sender_id (str): The unique identifier of the agent sending the message. This is typically your agent’s ID. (default: :obj:"system")
• exclude_agents (Optional[List[str]]): A list of agent IDs to exclude from the broadcast. The sender is automatically excluded. (default: :obj:None)

​

get_message_history

def get_message_history(self, agent_id: str, limit: Optional[int] = None):

• agent_id (str): The unique identifier of the agent whose message history you want to retrieve. Use list_available_agents() to find available agent IDs.
• limit (Optional[int]): The maximum number of recent messages to return. If not specified, it will return all messages up to the system’s limit. (default: :obj:None)

​

get_conversation_thread

def get_conversation_thread(self, message_id: str):

• message_id (str): The unique identifier of any message within the conversation thread you want to retrieve.

​

list_available_agents

def list_available_agents(self):

​

remove_agent

def remove_agent(self, agent_id: str):

• agent_id (str): Unique identifier of the agent to remove.

​

get_toolkit_status

def get_toolkit_status(self):

​

_add_to_history

def _add_to_history(self, message: AgentMessage):

​

_update_conversation_thread

def _update_conversation_thread(self, parent_id: str, new_id: str):

​

get_tools

def get_tools(self):