Constructing Occasion-Pushed AI Brokers with UAgents and Google Gemini: A Modular Python Implementation Information

On this tutorial, we exhibit find out how to use the UAgents framework to construct a light-weight, event-driven AI agent structure on high of Google’s Gemini API. We’ll begin by making use of nest_asyncio to allow nested occasion loops, then configure your Gemini API key and instantiate the GenAI shopper. Subsequent, we’ll outline our communication contracts, Query and Reply Pydantic fashions, and spin up two UAgents: one “gemini_agent” that listens for incoming Query messages, invokes the Gemini “flash” mannequin to generate responses, and emits Reply messages; and one “client_agent” that triggers a question upon startup and handles the incoming reply. Lastly, we’ll learn to run these brokers concurrently utilizing Python’s multiprocessing utility and gracefully shut down the occasion loop as soon as the change is full, illustrating UAgents’ seamless orchestration of inter-agent messaging.

!pip set up -q uagents google-genai

We set up the UAgents framework and the Google GenAI shopper library, offering the required tooling to construct and run your event-driven AI brokers with Gemini. The q flag runs the set up quietly, holding your pocket book output clear. Try the Pocket book right here

import os, time, multiprocessing, asyncio
import nest_asyncio  
from google import genai
from pydantic import BaseModel, Area
from uagents import Agent, Context


nest_asyncio.apply()

We arrange our Python setting by importing important modules, system utilities (os, time, multiprocessing, asyncio), nest_asyncio for enabling nested occasion loops (vital in notebooks), the Google GenAI shopper, Pydantic for schema validation, and core UAgents lessons. Lastly, nest_asyncio.apply() patches the occasion loop so you may run asynchronous UAgents workflows seamlessly in interactive environments. Try the Pocket book right here

os.environ["GOOGLE_API_KEY"] = "Use Your Personal API Key Right here"


shopper = genai.Consumer()

Right here we set our Gemini API key within the setting. Remember to substitute the placeholder along with your precise key, after which initialize the GenAI shopper, which can deal with all subsequent requests to Google’s Gemini fashions. This step ensures our agent has authenticated entry to generate content material by the API.

class Query(BaseModel):
    query: str = Area(...)


class Reply(BaseModel):
    reply: str = Area(...)

These Pydantic fashions outline the structured message codecs that our brokers will change with one another. The Query mannequin carries a single query string area, and the Reply mannequin carries a single reply string area. By utilizing Pydantic, we get automated validation and serialization of incoming and outgoing messages, guaranteeing that every agent at all times works with well-formed knowledge.

ai_agent = Agent(
    title="gemini_agent",
    seed="agent_seed_phrase",
    port=8000,
    endpoint=["http://127.0.0.1:8000/submit"]
)


@ai_agent.on_event("startup")
async def ai_startup(ctx: Context):
    ctx.logger.information(f"{ai_agent.title} listening on {ai_agent.tackle}")


def ask_gemini(q: str) -> str:
    resp = shopper.fashions.generate_content(
        mannequin="gemini-2.0-flash",
        contents=f"Reply the query: {q}"
    )
    return resp.textual content


@ai_agent.on_message(mannequin=Query, replies=Reply)
async def handle_question(ctx: Context, sender: str, msg: Query):
    ans = ask_gemini(msg.query)
    await ctx.ship(sender, Reply(reply=ans))

On this block, we instantiate the UAgents “gemini_agent” with a novel title, seed phrase (for deterministic identification), listening port, and HTTP endpoint for message submissions. We then register a startup occasion handler that logs when the agent is prepared, guaranteeing visibility into its lifecycle. The synchronous helper ask_gemini wraps the GenAI shopper name to Gemini’s “flash” mannequin. On the identical time, the @ai_agent.on_message handler deserializes incoming Query messages, invokes ask_gemini, and asynchronously sends again a validated Reply payload to the unique sender. Try the Pocket book right here

client_agent = Agent(
    title="client_agent",
    seed="client_seed_phrase",
    port=8001,
    endpoint=["http://127.0.0.1:8001/submit"]
)


@client_agent.on_event("startup")
async def ask_on_start(ctx: Context):
    await ctx.ship(ai_agent.tackle, Query(query="What's the capital of France?"))


@client_agent.on_message(mannequin=Reply)
async def handle_answer(ctx: Context, sender: str, msg: Reply):
    print("📨 Reply from Gemini:", msg.reply)
    # Use a extra sleek shutdown
    asyncio.create_task(shutdown_loop())


async def shutdown_loop():
    await asyncio.sleep(1)  # Give time for cleanup
    loop = asyncio.get_event_loop()
    loop.cease()

We arrange a “client_agent” that, upon startup, sends a Query to the gemini_agent asking for the capital of France, then listens for an Reply, prints the obtained response, and gracefully shuts down the occasion loop after a quick delay. Try the Pocket book right here

def run_agent(agent):
    agent.run()


if __name__ == "__main__":
    p = multiprocessing.Course of(goal=run_agent, args=(ai_agent,))
    p.begin()
    time.sleep(2)  


    client_agent.run()


    p.be part of()

Lastly, we outline a helper run_agent perform that calls agent.run(), then makes use of Python’s multiprocessing to launch the gemini_agent in its course of. After giving it a second to spin up, it runs the client_agent in the primary course of, blocking till the reply round-trip completes, and at last joins the background course of to make sure a clear shutdown.

In conclusion, with this UAgents-focused tutorial, we now have a transparent blueprint for creating modular AI companies that talk by way of well-defined occasion hooks and message schemas. You’ve seen how UAgents simplifies agent lifecycle administration, registering startup occasions, dealing with incoming messages, and sending structured replies, all with out boilerplate networking code. From right here, you may increase your UAgents setup to incorporate extra refined dialog workflows, a number of message sorts, and dynamic agent discovery.

Try the Pocket book right here. All credit score for this analysis goes to the researchers of this undertaking. Additionally, be at liberty to observe us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our Publication.

Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its reputation amongst audiences.