Communication Methods
Inter-Agent Communication in Crowe
The Agent class in Crowe includes built-in capabilities for direct and multi-agent communication, enabling agents to exchange information, request assistance, and coordinate workflows with minimal overhead.
These communication features form the foundation of complex, collaborative AI systems where specialization and teamwork are essential.
Why Communication Between Agents Matters
In a well-designed Crowe system, different agents take on specialized roles. Some may focus on gathering data, while others interpret results or make decisions. By allowing them to message each other directly or broadcast instructions to a group, Crowe enables smooth collaboration and efficient workflow orchestration.
Available Communication Methods
send_to
One-to-one request and response
Passing data from a data collector to an analyst
broadcast_to
Send one task to many agents in parallel
Gathering multiple viewpoints on a decision
handle_incoming
Process and respond to messages received from other agents
Acting on delegated work
dispatch_message
Send a structured, formatted message to a named agent
Sending notifications or system updates
Core Features
Direct Messaging – Simple one-to-one task assignment or consultation.
Parallel Broadcasting – Reach multiple agents and gather results in parallel.
Context-Aware Responses – Built-in contextual handling for incoming messages.
Error Tolerance – Graceful recovery from failed communications.
Threaded Execution – Parallel processing via
ThreadPoolExecutorfor speed.Flexible Parameters – Support for images, extra arguments, and keyword parameters.
Key Methods
send_to(agent, task, *args, img=None, **kwargs)
Initiates a direct interaction between the current agent and another agent.
Example:
from crowe import Agent
# Create a data acquisition agent
data_harvester = Agent(
agent_name="Crypto-Data-Harvester",
system_prompt=(
"You are responsible for fetching and structuring real-time cryptocurrency market data "
"from multiple exchanges, ensuring accuracy and formatting it for analysis."
),
max_loops=1,
)
# Create an AI-driven risk assessment agent
risk_analyzer = Agent(
agent_name="Risk-Analysis-Engine",
system_prompt=(
"You evaluate financial market conditions and assess the short-term and long-term risks "
"based on the structured crypto market data provided to you."
),
max_loops=1,
)
# Send a direct task from harvester to analyzer
risk_report = data_harvester.send_to(
agent=risk_analyzer,
task=(
"Using the latest BTC, ETH, and SOL market data from Binance and OKX, "
"analyze volatility levels, identify major risk factors, "
"and suggest protective strategies for a short-term investor."
)
)
print("===== RISK ANALYSIS REPORT =====")
print(risk_report)
broadcast_to(agent_list, task, *args, **kwargs)
Executes the same request across multiple agents concurrently and collects all responses.
Example:
from crowe import Agent
# Create a set of blockchain-focused agents
agents = [
Agent(
agent_name="OnChain-Data-Scout",
system_prompt="You specialize in gathering live on-chain transaction and wallet activity data from multiple blockchains.",
max_loops=1
),
Agent(
agent_name="Tokenomics-Strategist",
system_prompt="You analyze token supply, liquidity pools, and staking metrics to determine long-term sustainability.",
max_loops=1
),
Agent(
agent_name="Community-Sentiment-Tracker",
system_prompt="You monitor Twitter, Discord, and Telegram to assess community sentiment trends around specific crypto projects.",
max_loops=1
)
]
# Lead coordinator for multi-agent blockchain research
project_lead = Agent(
agent_name="DeFi-Research-Coordinator",
system_prompt="You coordinate blockchain research agents to deliver unified insights for Web3 investment decisions.",
max_loops=1
)
# Broadcast the research request to all agents
reports = project_lead.broadcast_to(
agent_list=agents,
task="Analyze the short-term growth potential of the $SOL token, "
"including on-chain activity, tokenomics health, and community sentiment."
)
# Present findings from each specialized agent
print("===== MULTI-AGENT RESEARCH REPORT =====")
for i, report in enumerate(reports, start=1):
print(f"[Agent {i}] {report or '⚠️ No data returned'}")
handle_incoming(from_agent, task, *args, **kwargs)
Handles an incoming message or task from another agent, optionally adding context.
Example:
from crowe import Agent
# Agent specialized in responding to cross-chain bridge issues
bridge_support = Agent(
agent_name="CrossChain-Support-Agent",
system_prompt="You resolve user issues related to cross-chain token transfers, stuck transactions, and gas fee discrepancies.",
max_loops=1
)
# Incoming request from transaction monitor agent
ticket_response = bridge_support.handle_incoming(
from_agent="Tx-Monitor-Agent",
task="A user’s SOL to ETH bridge transfer has been pending for over 2 hours. "
"Explain the likely cause and provide next steps to resolve the delay."
)
print("📨 Support Reply:")
print(ticket_response)
dispatch_message(target_name, message, *args, **kwargs)
Sends a structured notification to a named target agent.
Example:
from crowe import Agent
# Agent that handles blockchain event notifications
chain_notifier = Agent(
agent_name="OnChain-Event-Notifier",
system_prompt="You broadcast real-time blockchain event alerts to subscribed agents.",
max_loops=1
)
# Send an alert to the liquidity manager about a new pool creation
chain_notifier.dispatch_message(
target_name="Liquidity-Manager-Agent",
message="🚨 New Raydium liquidity pool detected: SOL/USDC — initial liquidity 500,000 USDC"
)
Summary
The Crowe Agent class offers a full-featured communication layer for building intelligent, distributed AI systems. Whether it’s a simple one-to-one message or a coordinated group broadcast, Crowe agents can collaborate seamlessly, handle context-rich interactions, and stay resilient to failures.
Last updated