Query Agents using proxy API

Overview

This file can be run on any platform supporting Python, with the necessary install permissions.

This example shows how to query an Agent using a proxy API.

The Agent

agent.py


 
from uagents import Agent, Context, Model
 
 
class TestRequest(Model):
    message: str
 
 
class Response(Model):
    text: str
 
 
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()}")
 
 
@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"))
 
 
if __name__ == "__main__":
    agent.run()

The Agent is created using the Agent class from uagents library. It is initialized 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 initialization. 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. For additional information on Agents’ handlers, check out this resource.

Proxy

proxy.py


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

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.

Run the example

In separate terminals:

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