Creating Modular Event-Driven AI Agents Using UAgents and Google Gemini in Python

Setting Up the Environment and Installing Dependencies

This tutorial demonstrates how to build a lightweight, event-driven AI agent architecture using the UAgents framework on top of Google's Gemini API. We begin by installing the necessary libraries:

!pip install -q uagents google-genai

The q flag ensures quiet installation, keeping outputs clean. This setup provides the tools needed to run event-driven AI agents with Gemini.

Preparing the Python Environment

We import essential modules and apply nest_asyncio to enable nested event loops, a necessity for running asynchronous workflows in interactive environments such as notebooks.

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()

The imports include system utilities, Google GenAI client, Pydantic for data validation, and core UAgents classes.

Configuring the Gemini API Client

Set your Gemini API key in the environment variables and initialize the GenAI client to authenticate requests for content generation.

os.environ["GOOGLE_API_KEY"] = "Use Your Own API Key Here"
 
client = genai.Client()

Replace the placeholder with your actual API key to enable access.

Defining Message Schemas with Pydantic

We define structured message formats for inter-agent communication using Pydantic models.

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

These models ensure that messages exchanged between agents are well-formed and validated.

Creating the Gemini Agent

The gemini_agent listens for incoming questions, invokes the Gemini "flash" model for answers, and sends back the responses.

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))

This agent registers startup and message handlers, integrating with Gemini to generate answers dynamically.

Creating the Client Agent

The client_agent initiates a question at startup and handles the incoming answer by printing it and shutting down gracefully.

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(" 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()

This agent demonstrates initiating communication and managing lifecycle events.

Running Agents Concurrently

We use Python's multiprocessing to run the agents concurrently, ensuring proper startup and shutdown sequences.

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()

The gemini_agent runs in a separate process while the client_agent runs in the main thread, allowing asynchronous communication between them.

Summary

This guide provides a modular blueprint for building event-driven AI agents using UAgents and Google Gemini. It covers environment setup, agent creation, message handling, and concurrent execution, enabling developers to create sophisticated AI workflows with minimal boilerplate.