UNDERSTANDING FETCH.AI AGENT CHAT PROTOCOL

4月之前•

Fetch.AI

看涨:

1

看跌:

0

共享

INTRODUCTION

Imagine a world where software agents can communicate with each other just like humans do through natural conversations. This is exactly what the Fetch.ai Agent Chat Protocol enables.

In this comprehensive guide, we’ll break down this powerful protocol that allows autonomous agents to chat, collaborate, and solve problems together.

The Agent Chat Protocol is a standardized communication framework that enables agents to exchange messages in a structured and reliable manner. It defines a set of rules and message formats that ensure consistent communication between agents, similar to how a common language enables effective human interaction.

Reference: https://innovationlab.fetch.ai/resources/docs/agent-communication/agent-chat-protocol

WHAT IS THE AGENT CHAT PROTOCOL?

The Agent Chat Protocol is like a universal language that all Fetch.ai agents can speak. Think of it as the “English” of the agent world no matter what type of agent you build, they can all communicate using this standardized protocol.

Key Concepts in Simple Terms:

Agents: Think of them as digital assistants that can perform tasks
Protocol: A set of rules that agents follow when talking to each other
Chat: Natural language conversations between agents
Messages: The actual words/data that agents exchange

HOW DOES IT WORK?

The Basic Flow:

User → Agent A → Agent B → Response

Here’s what happens:

A user sends a message to Agent A
Agent A processes the message and decides it needs help from Agent B
Agent A sends a chat message to Agent B
Agent B responds with helpful information
Agent A combines everything and responds to the user

Real-World Analogy:

Think of it like ordering food at a restaurant:

You = The customer (user)
Waiter = Agent A (receives your order)
Kitchen = Agent B (processes your request)
Menu = The chat protocol (standardized way to communicate)

The waiter doesn’t cook your food, but they know how to talk to the kitchen to get your order fulfilled.

CORE COMPONENTS

1. CORE MODELS

TextContent:
The basic content type for text messages. It uses Literal[‘text’] to ensure type safety.

from uagents_core.contrib.protocols.chat import TextContent

text_content = TextContent(
    type="text",
    text="Hello, can you help me with a task?"
)

Resource and ResourceContent:

Represents external resources like files, images, etc.

from uagents_core.contrib.protocols.chat import Resource, ResourceContent

resource = Resource(
    uri="https://example.com/file.pdf",
    metadata={"mime_type": "application/pdf", "role": "document"}
)

resource_content = ResourceContent(
    type="resource",
    resource_id=uuid4(),
    resource=resource
)

MetadataContent:

For sending metadata-only messages.

from uagents_core.contrib.protocols.chat import MetadataContent

metadata_content = MetadataContent(
    type="metadata",
    metadata={"mime_type": "text/plain", "role": "user_input"}
)

2. SESSION AND STREAM MANAGEMENT

Session Control:

Manages chat session lifecycle.

from uagents_core.contrib.protocols.chat import StartSessionContent, EndSessionContent

# Start a new conversation
start_session = StartSessionContent(type="start-session")

# End a conversation
end_session = EndSessionContent(type="end-session")

Stream Control:

Handles continuous data streams.

from uagents_core.contrib.protocols.chat import StartStreamContent, EndStreamContent

start_stream = StartStreamContent(
    type="start-stream",
    stream_id=uuid4()
)

end_stream = EndStreamContent(
    type="end-stream",
    stream_id=uuid4()
)

3. MESSAGE TYPES

ChatMessage:

The primary message type for communication.

from uagents_core.contrib.protocols.chat import ChatMessage, TextContent
from datetime import datetime
from uuid import uuid4

message = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[
        TextContent(type="text", text="Hello, can you help me with a task?")
    ]
)

What this does:

timestamp: When the message was sent (UTC)
msg_id: Unique identifier for the message
content: List of content elements (text, files, etc.)

ChatAcknowledgement:

Confirms message receipt.

from uagents_core.contrib.protocols.chat import ChatAcknowledgement

acknowledgement = ChatAcknowledgement(
    timestamp=datetime.utcnow(),
    acknowledged_msg_id=message.msg_id
)

4. MESSAGE HANDLERS

@protocol.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    print('I got a chat message', sender, msg)

@protocol.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    print('I got a chat acknowledgement', sender, msg)

BUILDING YOUR FIRST CHAT AGENT

Let’s create a simple agent that can chat with other agents.

Step 1: Basic Setup

from uagents import Agent, Context, Protocol
from uagents_core.contrib.protocols.chat import (
    ChatMessage, 
    ChatAcknowledgement,
    TextContent,
    StartSessionContent,
    EndSessionContent,
    chat_protocol_spec
)
from datetime import datetime
from uuid import uuid4

# Create your agent
my_agent = Agent(
    name="Helper Agent",
    seed="my_secret_seed_phrase",
    port=8000
)

# Initialize the chat protocol
chat_protocol = Protocol(spec=chat_protocol_spec)

Step 2: Handle Incoming Messages

@chat_protocol.on_message(ChatMessage)
async def handle_chat_message(ctx: Context, sender: str, msg: ChatMessage):
    """Handle incoming chat messages from other agents."""
    
    # Log the message
    ctx.logger.info(f"Received message from {sender}: {msg.content}")
    
    # Send acknowledgment
    await ctx.send(
        sender,
        ChatAcknowledgement(
            timestamp=datetime.utcnow(),
            acknowledged_msg_id=msg.msg_id
        )
    )
    
    # Process each piece of content in the message
    for item in msg.content:
        if isinstance(item, StartSessionContent):
            ctx.logger.info("New conversation started!")
            
        elif isinstance(item, TextContent):
            # Handle text messages
            response = await process_text_message(item.text)
            
            # Send response back
            response_msg = ChatMessage(
                timestamp=datetime.utcnow(),
                msg_id=uuid4(),
                content=[TextContent(type="text", text=response)]
            )
            await ctx.send(sender, response_msg)
            
        elif isinstance(item, EndSessionContent):
            ctx.logger.info("Conversation ended!")

Step 3: Process Messages

async def process_text_message(text: str) -> str:
    """Process incoming text messages and generate responses."""
    
    text = text.lower()
    
    if "hello" in text or "hi" in text:
        return "Hello! How can I help you today?"
    
    elif "weather" in text:
        return "I can help you get weather information. What city are you interested in?"
    
    elif "help" in text:
        return "I'm here to help! I can assist with weather, calculations, and more."
    
    else:
        return "I'm not sure how to help with that. Try asking for 'help' to see what I can do."

Step 4: Include the Protocol

# Add the chat protocol to your agent
my_agent.include(chat_protocol, publish_manifest=True)

# Run the agent
if __name__ == "__main__":
    my_agent.run()

COMPLETE EXAMPLE: TWO AGENTS COMMUNICATING

Agent1 Script (agent1.py):

from datetime import datetime
from uuid import uuid4
from uagents import Agent, Protocol, Context
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    chat_protocol_spec,
)

# Initialize agent1
agent1 = Agent()

# Store agent2's address (replace with actual address)
agent2_address = "agent1qf8n9q8ndlfvphmnwjzj9p077yq0m6kqc22se9g89y5en22sc38ck4p4e8d"

# Initialize the chat protocol
chat_proto = Protocol(spec=chat_protocol_spec)

# Startup Handler - Print agent details and send initial message
@agent1.on_event("startup")
async def startup_handler(ctx: Context):
    # Print agent details
    ctx.logger.info(f"My name is {ctx.agent.name} and my address is {ctx.agent.address}")
    
    # Send initial message to agent2
    initial_message = ChatMessage(
        timestamp=datetime.utcnow(),
        msg_id=uuid4(),
        content=[TextContent(type="text", text="Hello from Agent1!")]
    )
    
    await ctx.send(agent2_address, initial_message)

# Message Handler - Process received messages and send acknowledgements
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    for item in msg.content:
        if isinstance(item, TextContent):
            # Log received message
            ctx.logger.info(f"Received message from {sender}: {item.text}")
            
            # Send acknowledgment
            ack = ChatAcknowledgement(
                timestamp=datetime.utcnow(),
                acknowledged_msg_id=msg.msg_id
            )
            await ctx.send(sender, ack)
            
            # Send response message
            response = ChatMessage(
                timestamp=datetime.utcnow(),
                msg_id=uuid4(),
                content=[TextContent(type="text", text="Hello from Agent1!")]
            )
            await ctx.send(sender, response)

# Acknowledgement Handler - Process received acknowledgements
@chat_proto.on_message(ChatAcknowledgement)
async def handle_acknowledgement(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(f"Received acknowledgement from {sender} for message: {msg.acknowledged_msg_id}")

# Include the protocol in the agent
agent1.include(chat_proto, publish_manifest=True)

if __name__ == '__main__':
    agent1.run()

Agent2 Script (agent2.py):

from datetime import datetime
from uuid import uuid4
from uagents import Agent, Protocol, Context
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    chat_protocol_spec,
)

# Initialize agent2
agent2 = Agent()

# Initialize the chat protocol
chat_proto = Protocol(spec=chat_protocol_spec)

# Startup Handler - Print agent details
@agent2.on_event("startup")
async def startup_handler(ctx: Context):
    # Print agent details
    ctx.logger.info(f"My name is {ctx.agent.name} and my address is {ctx.agent.address}")

# Message Handler - Process received messages and send acknowledgements
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    for item in msg.content:
        if isinstance(item, TextContent):
            # Log received message
            ctx.logger.info(f"Received message from {sender}: {item.text}")
            
            # Send acknowledgment
            ack = ChatAcknowledgement(
                timestamp=datetime.utcnow(),
                acknowledged_msg_id=msg.msg_id
            )
            await ctx.send(sender, ack)
            
            # Send response message
            response = ChatMessage(
                timestamp=datetime.utcnow(),
                msg_id=uuid4(),
                content=[TextContent(type="text", text="Hello from Agent2!")]
            )
            await ctx.send(sender, response)

# Acknowledgement Handler - Process received acknowledgements
@chat_proto.on_message(ChatAcknowledgement)
async def handle_acknowledgement(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(f"Received acknowledgement from {sender} for message: {msg.acknowledged_msg_id}")

# Include the protocol in the agent
agent2.include(chat_proto, publish_manifest=True)

if __name__ == '__main__':
    agent2.run()

Running the Agents:

1. Start Agent2 first

2. Copy Agent2’s address from the startup logs and update it in Agent1’s script

3. Start Agent1

Expected Output:

When running both agents, you should see output similar to:

Agent2 Logs:

“My name is agent2 and my address is agent1qf8n9q8ndlfvphmnwjzj9p077yq0m6kqc22se9g89y5en22sc38ck4p4e8d”
“Received message from agent1: Hello from Agent1!”
“Received acknowledgement from agent1 for message: [message_id]”

Agent1 Logs:

“My name is agent1 and my address is [agent1_address]”
“Received message from agent2: Hello from Agent2!”
“Received acknowledgement from agent2 for message: [message_id]”

ADVANCED FEATURES

1. Multi-Agent Conversations

Agents can talk to multiple other agents simultaneously:

# Agent A talks to Agent B and Agent C
await ctx.send(agent_b_address, message_to_b)
await ctx.send(agent_c_address, message_to_c)

# Wait for responses
response_from_b = await ctx.receive(agent_b_address)
response_from_c = await ctx.receive(agent_c_address)

2. Structured Data Exchange

Agents can send more than just text:

from uagents_core.contrib.protocols.chat import FileContent

# Send a file
file_message = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[
        FileContent(
            type="file",
            filename="data.json",
            data=json.dumps({"key": "value"})
        )
    ]
)

3. Session Management

# Start a session
start_msg = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[StartSessionContent(type="start-session")]
)

# End a session
end_msg = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[EndSessionContent(type="end-session")]
)

REAL-WORLD USE CASES

1. Task Delegation

# Agent A receives a complex task
async def handle_complex_task(ctx: Context, task: str):
    # Break down the task
    subtasks = break_down_task(task)
    
    # Delegate to specialized agents
    for subtask in subtasks:
        specialized_agent = find_specialist(subtask)
        await ctx.send(specialized_agent, create_task_message(subtask))

2. Information Gathering

# Agent needs information from multiple sources
async def gather_information(ctx: Context, query: str):
    # Ask multiple agents for information
    agents = ["weather_agent", "news_agent", "stock_agent"]
    
    responses = []
    for agent in agents:
        response = await ctx.send(agent, create_query_message(query))
        responses.append(response)
    
    # Combine all responses
    return combine_responses(responses)

3. Collaborative Problem Solving

# Multiple agents work together to solve a problem
async def solve_problem(ctx: Context, problem: str):
    # Agent A: Analyzes the problem
    analysis = await ctx.send(analyst_agent, problem)
    
    # Agent B: Generates solutions
    solutions = await ctx.send(solver_agent, analysis)
    
    # Agent C: Evaluates solutions
    best_solution = await ctx.send(evaluator_agent, solutions)
    
    return best_solution

BEST PRACTICES

1. Always Acknowledge Messages

# Good practice
await ctx.send(sender, ChatAcknowledgement(
    timestamp=datetime.utcnow(),
    acknowledged_msg_id=msg.msg_id
))

2. Handle Errors Gracefully

try:
    response = await process_message(msg)
    await ctx.send(sender, response)
except Exception as e:
    error_msg = ChatMessage(
        timestamp=datetime.utcnow(),
        msg_id=uuid4(),
        content=[TextContent(type="text", text=f"Sorry, I encountered an error: {str(e)}")]
    )
    await ctx.send(sender, error_msg)

3. Use Meaningful Message IDs

# Include context in message IDs
msg_id = f"{agent_name}_{task_type}_{timestamp}"

4. Log Everything

ctx.logger.info(f"Received message from {sender}")
ctx.logger.info(f"Processing message: {msg.content}")
ctx.logger.info(f"Sending response to {sender}")

COMMON PATTERNS

1. Request-Response Pattern

# Agent A sends a request
request = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[TextContent(type="text", text="Can you help me with this?")]
)
await ctx.send(agent_b_address, request)

# Agent B responds
response = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[TextContent(type="text", text="Here's the answer!")]
)
await ctx.send(agent_a_address, response)

2. Broadcast Pattern

# Send the same message to multiple agents
agents = [agent1, agent2, agent3]
broadcast_msg = ChatMessage(
    timestamp=datetime.utcnow(),
    msg_id=uuid4(),
    content=[TextContent(type="text", text="Important announcement!")]
)

for agent in agents:
    await ctx.send(agent, broadcast_msg)

3. Chain Pattern

# Agent A → Agent B → Agent C
msg_a_to_b = ChatMessage(...)
response_b = await ctx.send(agent_b, msg_a_to_b)

msg_b_to_c = ChatMessage(...)
response_c = await ctx.send(agent_c, msg_b_to_c)

# Send final result back to A
final_result = ChatMessage(...)
await ctx.send(agent_a, final_result)

DEBUGGING TIPS

1. Enable Detailed Logging

import logging
logging.basicConfig(level=logging.DEBUG)

2. Add Message Tracking

# Track message flow
ctx.storage.set(f"message_{msg.msg_id}", {
    "sender": sender,
    "timestamp": datetime.utcnow().isoformat(),
    "content": str(msg.content)
})

3. Monitor Agent Health

@agent.on_interval(period=30.0)
async def health_check(ctx: Context):
    ctx.logger.info("Agent is healthy and running!")

BASIC MESSAGE FLOW

The protocol follows a simple request-response pattern with acknowledgments:

Agent A sends a ChatMessage to Agent B
Agent B sends a ChatAcknowledgement back to Agent A
Agent B can then send a ChatMessage response to Agent A
Agent A sends a ChatAcknowledgement back to Agent B

This ensures reliable communication and message delivery confirmation.

CONCLUSION

The Fetch.ai Agent Chat Protocol is a powerful tool that enables agents to communicate naturally and effectively. By understanding these concepts and patterns, you can build sophisticated multi-agent systems that work together to solve complex problems.

Key Takeaways:

Standardization: All agents can communicate using the same protocol
Flexibility: Support for text, files, and structured data
Reliability: Built-in acknowledgments and error handling
Scalability: Easy to add new agents to existing conversations

Next Steps:

Start with a simple chat agent
Add more complex message handling
Experiment with multi-agent conversations
Build specialized agents for specific tasks
Create agent networks for complex problem solving

The possibilities are endless when agents can communicate effectively with each other!

For more advanced topics and detailed documentation, visit:

Agent Chat Protocol | Innovation Lab Resources

This guide covers the fundamentals of the Fetch.ai Agent Chat Protocol. The examples and patterns shown here provide a solid foundation for building sophisticated multi-agent systems using the uAgents framework.

UNDERSTANDING FETCH.AI AGENT CHAT PROTOCOL was originally published in Fetch.ai on Medium, where people are continuing the conversation by highlighting and responding to this story.

4月之前•

Fetch.AI

看涨:

1

看跌:

0

共享

从同一位置管理所有加密资产、NFT 和 DeFi 资产

安全地关联您正在使用的投资组合，以开始交易。