Building Event-Driven AI Agents with UAgents and Google Gemini: A Modular Python Implementation Guide

In this tutorial, we demonstrate how to use the UAgents framework to build a lightweight, event-driven AI agent architecture on top of Google’s Gemini API. We’ll start by applying nest_asyncio to enable nested event loops, then configure your Gemini API key and instantiate the GenAI client. Next, we’ll define our communication contracts, Question and Answer Pydantic models, and spin up two UAgents: one “gemini_agent” that listens for incoming Question messages, invokes the Gemini “flash” model to generate responses, and emits Answer messages; and one “client_agent” that triggers a query upon startup and handles the incoming answer. Finally, we’ll learn how to run these agents concurrently using Python’s multiprocessing utility and gracefully shut down the event loop once the exchange is complete, illustrating UAgents’ seamless orchestration of inter-agent messaging.

!pip install -q uagents google-genai

We install the UAgents framework and the Google GenAI client library, providing the necessary tooling to build and run your event-driven AI agents with Gemini. The q flag runs the installation quietly, keeping your notebook output clean. Check out the Notebook here

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


nest_asyncio.apply()

We set up our Python environment by importing essential modules, system utilities (os, time, multiprocessing, asyncio), nest_asyncio for enabling nested event loops (critical in notebooks), the Google GenAI client, Pydantic for schema validation, and core UAgents classes. Finally, nest_asyncio.apply() patches the event loop so you can run asynchronous UAgents workflows seamlessly in interactive environments. Check out the Notebook here

os.environ["GOOGLE_API_KEY"] = "Use Your Own API Key Here"


client = genai.Client()

Here we set our Gemini API key in the environment. Be sure to replace the placeholder with your actual key, and then initialize the GenAI client, which will handle all subsequent requests to Google’s Gemini models. This step ensures our agent has authenticated access to generate content through the API.

class Question(BaseModel):
    question: str = Field(...)


class Answer(BaseModel):
    answer: str = Field(...)

These Pydantic models define the structured message formats that our agents will exchange with each other. The Question model carries a single question string field, and the Answer model carries a single answer string field. By using Pydantic, we get automatic validation and serialization of incoming and outgoing messages, ensuring that each agent always works with well-formed data.

ai_agent = Agent(
    name="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.info(f"{ai_agent.name} listening on {ai_agent.address}")


def ask_gemini(q: str) -> str:
    resp = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=f"Answer the question: {q}"
    )
    return resp.text


@ai_agent.on_message(model=Question, replies=Answer)
async def handle_question(ctx: Context, sender: str, msg: Question):
    ans = ask_gemini(msg.question)
    await ctx.send(sender, Answer(answer=ans))

In this block, we instantiate the UAgents “gemini_agent” with a unique name, seed phrase (for deterministic identity), listening port, and HTTP endpoint for message submissions. We then register a startup event handler that logs when the agent is ready, ensuring visibility into its lifecycle. The synchronous helper ask_gemini wraps the GenAI client call to Gemini’s “flash” model. At the same time, the @ai_agent.on_message handler deserializes incoming Question messages, invokes ask_gemini, and asynchronously sends back a validated Answer payload to the original sender. Check out the Notebook here

client_agent = Agent(
    name="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.send(ai_agent.address, Question(question="What is the capital of France?"))


@client_agent.on_message(model=Answer)
async def handle_answer(ctx: Context, sender: str, msg: Answer):
    print("<img src="https://s.w.org/images/core/emoji/15.1.0/72x72/1f4e8.png" alt="📨" class="wp-smiley" /> Answer from Gemini:", msg.answer)
    # Use a more graceful shutdown
    asyncio.create_task(shutdown_loop())


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

We set up a “client_agent” that, upon startup, sends a Question to the gemini_agent asking for the capital of France, then listens for an Answer, prints the received response, and gracefully shuts down the event loop after a brief delay. Check out the Notebook here

def run_agent(agent):
    agent.run()


if __name__ == "__main__":
    p = multiprocessing.Process(target=run_agent, args=(ai_agent,))
    p.start()
    time.sleep(2)  


    client_agent.run()


    p.join()

Finally, we define a helper run_agent function that calls agent.run(), then uses Python’s multiprocessing to launch the gemini_agent in its process. After giving it a moment to spin up, it runs the client_agent in the main process, blocking until the answer round-trip completes, and finally joins the background process to ensure a clean shutdown.

In conclusion, with this UAgents-focused tutorial, we now have a clear blueprint for creating modular AI services that communicate via well-defined event hooks and message schemas. You’ve seen how UAgents simplifies agent lifecycle management, registering startup events, handling incoming messages, and sending structured replies, all without boilerplate networking code. From here, you can expand your UAgents setup to include more sophisticated conversation workflows, multiple message types, and dynamic agent discovery.

Check out the Notebook here. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter.

The post Building Event-Driven AI Agents with UAgents and Google Gemini: A Modular Python Implementation Guide appeared first on MarkTechPost.

Source: Read More