Tutorials

Agent Chat Protocol

Overview

The Agent Chat Protocol enables agents to understand natural language and interoperate across ecosystems. With this protocol, an agent can receive user messages, acknowledge them, optionally request structured outputs from other agents, and reply back in a consistent format.

What you’ll build

This tutorial builds an agent that:

  • Receives chat messages in natural language
  • Acknowledges the message using the protocol
  • Requests a structured output (JSON) from another agent (an OpenAI-backed agent)
  • Uses the structured output to fetch current weather
  • Sends a natural language reply back to the user
Agent Chat ProtocolAgent Chat Protocol (Dark)

Prerequisites

  • Python 3.10+
  • Install dependencies:
pip install uagents requests

Step 1: Create an agent

Start with a minimal agent.

from uagents import Agent
agent = Agent()
  • The Agent is the main runtime that sends and receives protocol-compliant messages.

Step 2: Include the Chat Protocol

Attach the Agent Chat Protocol so the agent can send and receive ChatMessage and ChatAcknowledgement messages.

from uagents import Context, Protocol
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    chat_protocol_spec,
    StartSessionContent,
    EndSessionContent,
)
chat_proto = Protocol(spec=chat_protocol_spec)
  • Protocol(spec=chat_protocol_spec) adds handlers for a standard chat schema (text content, session start/end, acks).

Step 3: Include a Structured Output Client Protocol

To request structured data (JSON) from another agent, define a small client protocol and two models: one for the prompt and schema, one for the response.

from typing import Any, Dict
from uagents import Model
class StructuredOutputPrompt(Model):
    prompt: str
    output_schema: Dict[str, Any]
class StructuredOutputResponse(Model):
    output: Dict[str, Any]
struct_output_client_proto = Protocol(
    name="StructuredOutputClientProtocol", version="0.1.0"
)
  • StructuredOutputPrompt goes to a remote AI agent, which returns a StructuredOutputResponse carrying a JSON object that matches the schema.

Step 4: Create a text ChatMessage helper

A small helper to create consistent ChatMessage replies. Note the EndSessionContent uses type="end-session".

from datetime import datetime, timezone
from uuid import uuid4
def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
    content = [TextContent(type="text", text=text)]
    if end_session:
        content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=content,
    )

Step 5: Handle incoming chat and forward to the AI agent

  • Acknowledge each inbound ChatMessage
  • Forward the user’s text to an AI agent that can return structured output matching the weather request schema
from uagents import Agent, Context
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    StartSessionContent,
)
from functions import get_weather, WeatherRequest
AI_AGENT_ADDRESS = (
    "agent1qtlpfshtlcxekgrfcpmv7m9zpajuwu7d5jfyachvpa4u3dkt6k0uwwp2lct"  # OpenAI AI agent address
)
agent = Agent()
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    ctx.logger.info(f"Got a message from {sender}: {msg.content}")
    # Remember who sent this session's message so we can reply later
    ctx.storage.set(str(ctx.session), sender)
    # Acknowledge receipt
    await ctx.send(
        sender,
        ChatAcknowledgement(
            timestamp=datetime.now(timezone.utc), acknowledged_msg_id=msg.msg_id
        ),
    )
    # Extract user text and forward to AI agent for structured output
    for item in msg.content:
        if isinstance(item, StartSessionContent):
            ctx.logger.info(f"Got a start session message from {sender}")
            continue
        elif isinstance(item, TextContent):
            ctx.logger.info(f"User said: {item.text}")
            # Ask the AI agent to produce JSON matching our schema
            await ctx.send(
                AI_AGENT_ADDRESS,
                StructuredOutputPrompt(
                    prompt=item.text, output_schema=WeatherRequest.schema()
                ),
            )
        else:
            ctx.logger.info(f"Got unexpected content from {sender}")

Step 6: Handle structured output response and reply

  • When the remote AI agent replies with a JSON object, parse it, fetch the weather, and send a natural-language message back to the original user. The final formatting happens on the agent side.
@struct_output_client_proto.on_message(StructuredOutputResponse)
async def handle_structured_output_response(
    ctx: Context, sender: str, msg: StructuredOutputResponse
):
    ctx.logger.info(f"Structured output: {msg.output}")
    # Who started this session?
    session_sender = ctx.storage.get(str(ctx.session))
    if session_sender is None:
        ctx.logger.error("No session sender found in storage")
        return
    # Handle unknowns gracefully
    if "<UNKNOWN>" in str(msg.output):
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't process your location request. Please try again later."
            ),
        )
        return
    # Extract location from structured output
    try:
        location = msg.output.get("location") if isinstance(msg.output, dict) else None
    except Exception:
        location = None
    try:
        if not location:
            raise ValueError("No location provided in structured output")
        weather = get_weather(location)
        ctx.logger.info(str(weather))
    except Exception as err:
        ctx.logger.error(f"Error: {err}")
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't process your request. Please try again later."
            ),
        )
        return
    # Reply uses pre-formatted text from get_weather
    reply = weather.get("weather") or f"Weather for {location}: (no data)"
    await ctx.send(session_sender, create_text_chat(reply))

Step 7: Wire up protocols and run

agent.include(chat_proto, publish_manifest=True)
agent.include(struct_output_client_proto, publish_manifest=True)
if __name__ == "__main__":
    agent.run()

Weather utility module (functions.py)

This helper module defines the expected schema for the structured output and a function to fetch weather from Open-Meteo. It returns a single pre-formatted string under weather.

The full module is in Complete example below, alongside the finished agents.py.

Complete example (copy-paste)

Combine everything into two files you can run as a hosted agent or locally.

This Weather Agent is an Agentverse hosted agent. You can create your own hosted agent by following the guide here:

Hosted Agents

.

# agents.py
from uagents import Agent, Context, Protocol
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    chat_protocol_spec,
    StartSessionContent,
    EndSessionContent,
)
from functions import get_weather, WeatherRequest
from datetime import datetime, timezone
from uuid import uuid4
from typing import Any, Dict
from uagents import Model
class StructuredOutputPrompt(Model):
    prompt: str
    output_schema: Dict[str, Any]
class StructuredOutputResponse(Model):
    output: Dict[str, Any]
AI_AGENT_ADDRESS = "agent1qtlpfshtlcxekgrfcpmv7m9zpajuwu7d5jfyachvpa4u3dkt6k0uwwp2lct" # OpenAI ai agent address
agent = Agent()
chat_proto = Protocol(spec=chat_protocol_spec)
struct_output_client_proto = Protocol(
    name="StructuredOutputClientProtocol", version="0.1.0"
)
def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
    content = [TextContent(type="text", text=text)]
    if end_session:
        content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=content,
    )
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    ctx.logger.info(f"Got a message from {sender}: {msg.content}")
    ctx.storage.set(str(ctx.session), sender)
    await ctx.send(
        sender,
        ChatAcknowledgement(timestamp=datetime.now(timezone.utc), acknowledged_msg_id=msg.msg_id),
    )
    for item in msg.content:
        if isinstance(item, StartSessionContent):
            ctx.logger.info(f"Got a start session message from {sender}")
            continue
        elif isinstance(item, TextContent):
            ctx.logger.info(f"Got a message from {sender}: {item.text}")
            ctx.storage.set(str(ctx.session), sender)
            await ctx.send(
                AI_AGENT_ADDRESS,
                StructuredOutputPrompt(
                    prompt=item.text, output_schema=WeatherRequest.schema()
                ),
            )
        else:
            ctx.logger.info(f"Got unexpected content from {sender}")
@chat_proto.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(
        f"Got an acknowledgement from {sender} for {msg.acknowledged_msg_id}"
    )
@struct_output_client_proto.on_message(StructuredOutputResponse)
async def handle_structured_output_response(
    ctx: Context, sender: str, msg: StructuredOutputResponse
):
    ctx.logger.info(f'Here is the message from structured output {msg.output}')
    session_sender = ctx.storage.get(str(ctx.session))
    if session_sender is None:
        ctx.logger.error(
            "Discarding message because no session sender found in storage"
        )
        return
    if "<UNKNOWN>" in str(msg.output):
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't process your location request. Please try again later."
            ),
        )
        return
    # Extract location robustly from dict output
    try:
        location = msg.output.get("location") if isinstance(msg.output, dict) else None
    except Exception:
        location = None
    try:
        if not location:
            raise ValueError("No location provided in structured output")
        weather = get_weather(location)
        ctx.logger.info(str(weather))
    except Exception as err:
        ctx.logger.error(f"Error: {err}")
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't process your request. Please try again later."
            ),
        )
        return
    reply = weather.get("weather") or f"Weather for {location}: (no data)"
    chat_message = create_text_chat(reply)
    await ctx.send(session_sender, chat_message)
agent.include(chat_proto, publish_manifest=True)
agent.include(struct_output_client_proto, publish_manifest=True)
if __name__ == "__main__":
    agent.run()
# functions.py
from uagents import Model
import requests
class WeatherRequest(Model):
    location : str
class WeatherResponse(Model):
    weather : str
def get_weather(location: str):
    """Return current weather for a location string (e.g., 'Paris, France')."""
    if not location or not location.strip():
        raise ValueError("location is required")
    # 1) Geocode
    geo_params = {"name": location, "count": 1, "language": "en", "format": "json"}
    gr = requests.get(
        "https://geocoding-api.open-meteo.com/v1/search",
        params=geo_params,
        timeout=60,
    )
    gr.raise_for_status()
    g = gr.json()
    if not g.get("results"):
        raise RuntimeError(f"No geocoding match for: {location}")
    r0 = g["results"][0]
    latitude = r0["latitude"]
    longitude = r0["longitude"]
    timezone = r0.get("timezone") or "auto"
    display = ", ".join([v for v in [r0.get("name"), r0.get("admin1"), r0.get("country")] if v])
    # 2) Current weather
    wx_params = {
        "latitude": latitude,
        "longitude": longitude,
        "timezone": timezone,
        "current": (
            "temperature_2m,apparent_temperature,relative_humidity_2m,"
            "weather_code,wind_speed_10m,wind_direction_10m,is_day,precipitation"
        ),
    }
    wr = requests.get("https://api.open-meteo.com/v1/forecast", params=wx_params, timeout=60)
    wr.raise_for_status()
    data = wr.json()
    current = data.get("current") or data.get("current_weather") or {}
    temp = current.get("temperature_2m")
    app = current.get("apparent_temperature")
    wind = current.get("wind_speed_10m")
    rh = current.get("relative_humidity_2m")
    parts = [f"Weather for {display}"]
    if temp is not None:
        parts.append(f"temp {temp}°C")
    if app is not None:
        parts.append(f"feels like {app}°C")
    if rh is not None:
        parts.append(f"RH {rh}%")
    if wind is not None:
        parts.append(f"wind {wind} km/h")
    return {"weather": ", ".join(parts)}

Why the Agent Chat Protocol is useful

  • Natural language first: users can speak or type naturally; your agent wraps messages in a standard structure
  • Interoperability: any agent implementing the protocol can communicate, regardless of internal implementation
  • Extensible: add client protocols (like structured output) to connect to specialized agents
  • Reliability: acknowledgements and session controls make delivery and session state explicit

You can also use this agent through ASI:One by mentioning it directly in your prompt, for example:

@agent1qde95qr0dzcnhhs8f65hkwujn9mh89jx0u7u7g6nv3tm2jxvjwhkunvessq please get me weather of San Francisco.

To try a live conversation experience, visit the Example Weather Agent on Agentverse.

Agent chat in ASI:One (Light)Agent chat in ASI:One (Dark)

Next steps

  1. Planner Mode - How ASI:One discovers and calls agents like this one
  2. Chat Completions API - Call ASI:One directly from your own code
  3. Structured Data - Constraining an ASI:One reply to a JSON schema, the HTTP equivalent of the pattern above