fetch-logo
ConceptsConceptsGuidesGuidesExamplesExamplesReferencesReferencesAPIsAPIs
GitHub (opens in a new tab)
  • Guides
      • Quickstart
        • What's an Agent?
        • uAgents Framework installation
        • Create your first agent
        • Agents address
        • Communicating with other agents
        • Agent Handlers (on_...)
        • Agents storage functions
        • Public and private Agents
        • Send tokens with Agents
        • Agents Mailboxes
        • Agent Functions
        • Make agents AI Engine compatible
        • Multi-file agent pipeline for AI Engine: Hugging face API to create a multi agent pipeline
        • Options for running local agents
        • Hosted agent
        • Introducing dialogues
        • Almanac Contract
        • Verify messages with Agents
        • Multi-file agent pipeline for AI Engine: Network of Primary and Secondary functions in Agentverse
        • Localwallet
        • Agentverse: Hosted Agents
        • Agentverse: Dice Roll agent
        • Agentverse: Allowed Imports
        • Agentverse: Mailbox
        • Register Agentverse Functions
        • Agentverse Functions: coin toss agent
        • Field descriptions for DeltaV
        • Agentverse Command Line Interface (AVCTL)
        • AVCTL Hosting commands
      • Agents and Functions creation APIs
      • Secret Management APIs
      • How to convert native FET to and from ERC-20 FET
      • How to stake FET tokens
      • Different ways of staking FET
      • Re-delegating staked FET token
      • Reconciliation service
      • How to setup a Multisig Wallet
        • Getting started
        • How to use the Fetch wallet
        • How to stake and claim rewards
        • Address book
        • Connections
        • Fetch Wallet Hardware Connection Guide
        • Installation
        • Connecting to a blockchain
        • Querying balances
        • Wallets and private keys
        • Sending funds
        • Staking
        • Smart contracts
          • Stake auto-compounder
          • Stake optimizer
          • Oracles
          • Wallet top-up
          • Liquidity pool
          • Swap automation
        • Installation️
        • Getting started
        • Keys
          • How to add profiles
          • How to add contract templates
          • How to compile contracts
          • How to deploy contracts
          • Contract interaction
        • Installation
        • How to use chain state snapshots
        • State-synchronization (state-sync)
        • How to set up a validator node
        • How to join a testnet
        • How to run a single node test network
        • Governance
        • How to get testnet tokens via the Token Faucet
          • CLI - Introduction
          • CLI - Managing keys
          • CLI - Managing tokens
          • CLI - Multisig keys
          • CLI - Delegation
          • Governance proposals
      • Agents 101
      • Agents 101 for AI Engine
  • Guides
      • Quickstart
        • What's an Agent?
        • uAgents Framework installation
        • Create your first agent
        • Agents address
        • Communicating with other agents
        • Agent Handlers (on_...)
          • Introduction
          • Creating an interval task with on_interval() handler
          • Handle messages using the on_message() handler
          • Answer queries with on_query() handler
        • Agents storage functions
        • Public and private Agents
        • Send tokens with Agents
        • Agents Mailboxes
        • Agent Functions
        • Make agents AI Engine compatible
        • Multi-file agent pipeline for AI Engine: Hugging face API to create a multi agent pipeline
        • Options for running local agents
        • Hosted agent
        • Introducing dialogues
        • Almanac Contract
        • Verify messages with Agents
        • Multi-file agent pipeline for AI Engine: Network of Primary and Secondary functions in Agentverse
        • Localwallet
        • Agentverse: Hosted Agents
        • Agentverse: Dice Roll agent
        • Agentverse: Allowed Imports
        • Agentverse: Mailbox
        • Register Agentverse Functions
        • Agentverse Functions: coin toss agent
        • Field descriptions for DeltaV
        • Agentverse Command Line Interface (AVCTL)
        • AVCTL Hosting commands
      • Agents and Functions creation APIs
      • Secret Management APIs
      • How to convert native FET to and from ERC-20 FET
      • How to stake FET tokens
      • Different ways of staking FET
      • Re-delegating staked FET token
      • Reconciliation service
      • How to setup a Multisig Wallet
        • Getting started
        • How to use the Fetch wallet
        • How to stake and claim rewards
        • Address book
        • Connections
        • Fetch Wallet Hardware Connection Guide
        • Installation
        • Connecting to a blockchain
        • Querying balances
        • Wallets and private keys
        • Sending funds
        • Staking
        • Smart contracts
          • Stake auto-compounder
          • Stake optimizer
          • Oracles
          • Wallet top-up
          • Liquidity pool
          • Swap automation
        • Installation️
        • Getting started
        • Keys
          • How to add profiles
          • How to add contract templates
          • How to compile contracts
          • How to deploy contracts
          • Contract interaction
        • Installation
        • How to use chain state snapshots
        • State-synchronization (state-sync)
        • How to set up a validator node
        • How to join a testnet
        • How to run a single node test network
        • Governance
        • How to get testnet tokens via the Token Faucet
          • CLI - Introduction
          • CLI - Managing keys
          • CLI - Managing tokens
          • CLI - Multisig keys
          • CLI - Delegation
          • Governance proposals
      • Agents 101
      • Agents 101 for AI Engine
Guides
AI Agents
Intermediate topics
Agent Handlers (on_...)

Agent Handlers

Introduction

Within the uAgents Framework, you can allow agents to perform specific actions at the moment some sort of condition has been satisfied. You can do so, you will need to use a specific handler. These are like decorators and are identified by the following code syntax: .on_....

Below, we show how to use the following different event handlers:

  1. Interval tasks: .on_interval()
  2. Handle messages: .on_message()
  3. Answer queries: .on_query()

Creating an interval task with on_interval() handler

Sometimes an agent will need to perform a task periodically. To do this we can use the on_interval() decorator which periodically repeats a given function for the agent. For instance, an agent could send a message every 2 seconds to another agent.

Let's get started and create our first interval task!

Walk-through

  1. Let's create a Python script for this task, and name it: interval-task.py

  2. Then import the necessary classes from uagents library, Agent and Context, and create our agent:

    from uagents import Agent, Context
     
    agent = Agent(name="alice", seed="alice recovery phrase")
  3. Create a function to handle the startup event, which will introduce the agent:

    @agent.on_event("startup")
    async def introduce_agent(ctx: Context):
        ctx.logger.info(f"Hello, I'm agent {agent.name} and my address is {agent.address}.")
  4. We can now define our agent's interval behavior. We want our agent to log a message every 2 seconds using the on_interval decorator:

    @agent.on_interval(period=2.0)
    async def say_hello(ctx: Context):
        ctx.logger.info("Hello!")
     
    if __name__ == "__main__":
        agent.run()

    The output will be printed out using the ctx.logger.info() method.

  5. Save the script.

The overall script should look as follows:

interval-task.py
from uagents import Agent, Context
 
agent = Agent(name="alice", seed="alice recovery phrase")
 
@agent.on_event("startup")
async def introduce_agent(ctx: Context):
    ctx.logger.info(f"Hello, I'm agent {agent.name} and my address is {agent.address}.")
 
@agent.on_interval(period=2.0)
async def say_hello(ctx: Context):
    ctx.logger.info("Hello!")
 
if __name__ == "__main__":
    agent.run()

Run the script

Run the script: python interval-task.py

The output should be as follows:

hello, my name is alice
hello, my name is alice
hello, my name is alice

Handle messages using the on_message() handler

We now showcase a scenario where three agents, named alice, bob, and charles, use a custom protocol ↗️ to communicate. In the example, Alice and Bob support the protocol, whereas Charles attempts to send broadcast messages to all agents using the protocol. Agents use the on_message() handler which allows them to handle messages matching specific data models.

Let's get started!

Walk-through

  1. First of all, let's create a Python script for this task, and name it: broadcast.py

  2. We then need to import the Agent, Bureau, Context, Model, and Protocol classes from the uagents library, and the fund_agent_if_low from uagents.setup. Then, let's create the 3 different agents using the class Agent. Each agent is initialized with a unique name and a seed phrase for wallet recovery. Additionally, if an agent's wallet balance is low, the fund_agent_if_low() function is called to add funds to their wallet:

    from uagents import Agent, Bureau, Context, Model, Protocol
    from uagents.setup import fund_agent_if_low
     
    alice = Agent(name="alice", seed="alice recovery phrase")
    bob = Agent(name="bob", seed="bob recovery phrase")
    charles = Agent(name="charles", seed="charles recovery phrase")
     
    fund_agent_if_low(alice.wallet.address())
    fund_agent_if_low(bob.wallet.address())
    fund_agent_if_low(charles.wallet.address())

    It is optional but useful to include a seed parameter when creating an agent to set fixed addresses ↗️. Otherwise, random addresses will be generated every time you run the agent.

  3. Let's then define the message data models to specify the type of messages being handled and exchanged by the agents. We define a BroadcastExampleRequest and a BroadcastExampleResponse data models. Finally, create a protocol named proto with version 1.0:

    class BroadcastExampleRequest(Model):
        pass
     
    class BroadcastExampleResponse(Model):
        text: str
     
    proto = Protocol(name="proto", version="1.0")
  4. Let's now define a message handler function for incoming messages of type BroadcastExampleRequest in the protocol:

    @proto.on_message(model=BroadcastExampleRequest, replies=BroadcastExampleResponse)
    async def handle_request(ctx: Context, sender: str, _msg: BroadcastExampleRequest):
        await ctx.send(
            sender, BroadcastExampleResponse(text=f"Hello from {ctx.agent.name}")
        )

    Here we defined a handle_request() function which is used whenever a request is received. This sends a response back to the sender. This function is decorated with the .on_message() decorator indicating that this function is triggered whenever a message of type BroadcastExampleRequest is received. The function sends a response containing a greeting message with the name of the agent that sent the request in the first place.

  5. Now, we need to include the protocol into the agents. Specifically, the protocol is included in both alice and bob agents. This means they will follow the rules defined in the protocol when communicating:

    alice.include(proto)
    bob.include(proto)
    ℹ️

    After the first registration in the Almanac ↗️ smart contract, it will take about 5 minutes before the agents can be found through the protocol.

  6. It is now time to define the behavior and function of charles agent:

    @charles.on_interval(period=5)
    async def say_hello(ctx: Context):
        status_list = await ctx.broadcast(proto.digest, message=BroadcastExampleRequest())
        ctx.logger.info(f"Trying to contact {len(status_list)} agents.")
     
    @charles.on_message(model=BroadcastExampleResponse)
    async def handle_response(ctx: Context, sender: str, msg: BroadcastExampleResponse):
        ctx.logger.info(f"Received response from {sender}: {msg.text}")

    In the first part, we use the .on_interval() decorator to define an interval behavior for this agent when the script is being run. In this case, the agent will execute the say_hello() function every 5 seconds. The Context object is a collection of data and functions related to the agent. Inside the say_hello() function, the agent uses the ctx.broadcast() method to send a broadcast message. The message is of type BroadcastExampleRequest() and it is being sent using the protocol's digest (proto.digest).

    Then, we defined a .on_message() decorator which decorates handle_response() function. This function handles all incoming messages of type BroadcastExampleResponse from other agents. When a response is received, it logs the information. Inside the handle_response() function, the agent logs an informational message using ctx.logger.info() method to print the sender and the content of the message. The message includes the sender's name and the text content of the response message.

  7. We are now ready to set up a Bureau object for agents to be run together at the same time, and we add alice, bob, and charles to it using the bureau.add() method:

    bureau = Bureau(port=8000, endpoint="http://localhost:8000/submit")
    bureau.add(alice)
    bureau.add(bob)
    bureau.add(charles)
     
    if __name__ == "__main__":
        bureau.run()

    The bureau is assigned to listen on port=8000 and specifies an endpoint at "http://localhost:8000/submit" for submitting data.

  8. Save the script.

The overall script should look as follows:

broadcast.py
from uagents import Agent, Bureau, Context, Model, Protocol
 
# create agents
# alice and bob will support the protocol
# charles will try to reach all agents supporting the protocol
alice = Agent(name="alice", seed="alice recovery phrase")
bob = Agent(name="bob", seed="bob recovery phrase")
charles = Agent(name="charles", seed="charles recovery phrase")
 
class BroadcastExampleRequest(Model):
    pass
 
class BroadcastExampleResponse(Model):
    text: str
 
# define protocol
proto = Protocol(name="proto", version="1.0")
 
@proto.on_message(model=BroadcastExampleRequest, replies=BroadcastExampleResponse)
async def handle_request(ctx: Context, sender: str, _msg: BroadcastExampleRequest):
    await ctx.send(
        sender, BroadcastExampleResponse(text=f"Hello from {ctx.agent.name}")
    )
 
# include protocol
# Note: after the first registration on the almanac smart contract, it will
# take about 5 minutes before the agents can be found through the protocol
alice.include(proto)
bob.include(proto)
 
# let charles send the message to all agents supporting the protocol
@charles.on_interval(period=5)
async def say_hello(ctx: Context):
    status_list = await ctx.broadcast(proto.digest, message=BroadcastExampleRequest())
    ctx.logger.info(f"Trying to contact {len(status_list)} agents.")
 
@charles.on_message(model=BroadcastExampleResponse)
async def handle_response(ctx: Context, sender: str, msg: BroadcastExampleResponse):
    ctx.logger.info(f"Received response from {sender}: {msg.text}")
 
bureau = Bureau(port=8000, endpoint="http://localhost:8000/submit")
bureau.add(alice)
bureau.add(bob)
bureau.add(charles)
 
if __name__ == "__main__":
    bureau.run()

Run the script

Make sure to have activated your virtual environment correctly.

Run the script: python broadcast.py

The output would be:

Trying to contact 2 agents.
Received response from alice: Hello from alice
Received response from bob: Hello from bob

Answer queries with on_query() handler

The on_query() handler is used to register a Function ↗️ as a handler for incoming queries that match a specified Model. This decorator enables the agent to respond to queries in an event-driven manner.

Walk-through

Agent's script

For the agent, the script sets up an agent to handle incoming queries. It defines two models: TestRequest and Response. Upon startup, it logs the agent's details. The core functionality lies in the query_handler, decorated with @agent.on_query(), which processes received queries and sends back a predefined response. This demonstrates creating responsive agents within the uagents Framework, showcasing how they can interact with other agents or functions in an asynchronous, event-driven architecture.

agent.py
from uagents import Agent, Context, Model
 
class TestRequest(Model):
    message: str
 
class Response(Model):
    text: str
 
# Initialize the agent with its configuration.
agent = Agent(
    name="your_agent_name_here",
    seed="your_agent_seed_here",
    port=8001,
    endpoint="http://localhost:8001/submit",
)
 
@agent.on_event("startup")
async def startup(ctx: Context):
    ctx.logger.info(f"Starting up {agent.name}")
    ctx.logger.info(f"With address: {agent.address}")
    ctx.logger.info(f"And wallet address: {agent.wallet.address()}")
 
# Decorator to handle incoming queries.
@agent.on_query(model=TestRequest, replies={Response})
async def query_handler(ctx: Context, sender: str, _query: TestRequest):
    ctx.logger.info("Query received")
    try:
        # do something here
        await ctx.send(sender, Response(text="success"))
    except Exception:
        await ctx.send(sender, Response(text="fail"))
 
 
# Main execution block to run the agent.
if __name__ == "__main__":
    agent.run()

The agent is created using the Agent class from uagents library. It is initialised with a name, seed, port, and endpoint. It defines an on_event() handler for the startup event, where it logs information about the agent's initialisation. It defines an on_query() handler for handling queries of type TestRequest. Upon receiving a query, it processes it and sends back a Response. The agent is then set to run.

Proxy

The proxy is implemented using FastAPI. It sets up two routes: "/" for a simple root message and "/endpoint" for receiving requests. When a POST request is made to "/endpoint" with a JSON payload containing a TestRequest, it triggers the make_agent_call function. Inside make_agent_call, it calls agent_query to communicate with the agent. The agent receives the query, processes it, and sends back a response. The proxy receives the response from the agent and sends back a success message along with the response text.

Let's explore the Proxy code script step-by-step:

  1. First of all navigate to directory where you want to create your project.

  2. Create a Python script name on_query.py by running touch on_query.py.

  3. We need to import json, fastapi, uagent's Model and query. Then we would need to define the query format using the TestRequest class as a subclass of Model:

    import json
     
    from fastapi import FastAPI
    from uagents import Model
    from uagents.query import query
     
    AGENT_ADDRESS = "agent1qt6ehs6kqdgtrsduuzslqnrzwkrcn3z0cfvwsdj22s27kvatrxu8sy3vag0"
     
    class TestRequest(Model):
        message: str
  4. Create agent_query() function to send query to agent and decode the response received.

    async def agent_query(req):
        response = await query(destination=AGENT_ADDRESS, message=req, timeout=15.0)
        data = json.loads(response.decode_payload())
        return data["text"]
  5. Initialize a FastAPI app:

    app = FastAPI()
  6. Define a root endpoint to test the server:

    @app.get("/")
    def read_root():
        return "Hello from the Agent controller"
  7. Define an endpoint to make agent calls:

    @app.post("/endpoint")
    async def make_agent_call(req: TestRequest):
        try:
            res = await agent_query(req)
            return f"successful call - agent response: {res}"
        except Exception:
            return "unsuccessful agent call"
  8. Save the script.

The overall script should look as follows:

on_query.py
import json
 
from fastapi import FastAPI
from uagents import Model
from uagents.query import query
 
AGENT_ADDRESS = "agent1qt6ehs6kqdgtrsduuzslqnrzwkrcn3z0cfvwsdj22s27kvatrxu8sy3vag0"
 
class TestRequest(Model):
    message: str
 
async def agent_query(req):
    response = await query(destination=AGENT_ADDRESS, message=req, timeout=15.0)
    data = json.loads(response.decode_payload())
    return data["text"]
 
app = FastAPI()
 
@app.get("/")
def read_root():
    return "Hello from the Agent controller"
 
@app.post("/endpoint")
async def make_agent_call(req: TestRequest):
    try:
        res = await agent_query(req)
        return f"successful call - agent response: {res}"
    except Exception:
        return "unsuccessful agent call"

Run the example

In separate terminals:

  1. Run the FastAPI proxy: uvicorn proxy:app

  2. Run the agent: python agent.py

  3. Query the agent via the proxy: curl -d '{"message": "test"}' -H "Content-Type: application/json" -X POST http://localhost:8000/endpoint

Was this page helpful?

Communicating with other agentsAgents storage functions
footer-logo

Main website

Integrations

Events

We’re hiring!

Twitter (opens in a new tab)Telegram (opens in a new tab)Discord (opens in a new tab)GitHub (opens in a new tab)Youtube (opens in a new tab)LinkedIn (opens in a new tab)Reddit (opens in a new tab)
Sign up for developer updates