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

Copy CodeCopiedUse a different Browser

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("
                            
                                read more                                
                            
                        

Tags:

Previous Article

Mistral AI Releases Mistral Small 3.2: Enhanced Instruction Following, Reduced R...

Next Article

Why Generalization in Flow Matching Models Comes from Approximation, Not Stochas...

Related Posts

Highlighted at CVPR 2025: Google DeepMind’s ‘Motion Pro...

Jun 14, 2025  0

Sakana AI Introduces Text-to-LoRA (T2L): A Hypernetwork...

Jun 14, 2025  0

NVIDIA AI Introduces Fast-dLLM: A Training-Free Framewo...

Jun 2, 2025  0

Comments

Name

Email

Comment

Post Comment