Modular Python Guide for Building AI Agents that are Event Driven with UAgents & Google Gemini

This tutorial will show you how to create a. UAgents The framework is used to develop a lightweight AI agent on Google’s Gemini API. Nest_asyncio will be used to create event loops that are nested. Then, configure your Gemini API Key and the GenAI Client. We’ll then define the communication contracts and Question and Answer Pydantic Models, as well as spin up two UAgents. “gemini_agent” Gemini is a Gemini that monitors incoming Question messages. “flash” Answer message model that emits Answer messages and generates responses “client_agent” This code triggers an initial query and then handles any incoming response. We’ll also learn to use Python’s Multiprocessing tool and gracefully close the event loop after the exchange has been completed. This will demonstrate UAgents seamless orchestration for inter-agent communication.

Google-genai Agents:!pip Install -q

The UAgents framework is installed, along with the Google GenAI library. This provides all the required tools to develop and run AI event agents in Gemini. This flag will run the installation in a quiet manner, so that your notebook output is clean. Take a look at the Notebook Here is a link to the article

import os, time, multiprocessing, asyncio
import nest_asyncio  
From Google, import Genai
BaseModel Field
Import Agents from Uagents, Context


nest_asyncio.apply()

We import essential Python modules. These include the Google GenAI Client, Pydantic to validate schemas and core UAgents. Finally, nest_asyncio.apply() Patches the event loop to run asynchronous UAgents in an interactive environment. Take a look at the Notebook Here is a link to the article

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


Client = client()

We now set the Gemini API key. The placeholder should be replaced with your key. Initialize the GenAI Client, which will then handle any subsequent requests made to Google Gemini Models. This will ensure that the agent can generate content via API.

This class is a base model for the question (basis) type.
    question: str = Field(...)


Class Answer (BaseModel:
 The answer is: str = Field (...)

The Pydantic model defines the message structures that agents exchange. Each model has a question and answer field. Pydantic allows us to validate and serialize incoming and egressing messages automatically, so that agents always work with properly-formatted 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}"
    )
 Text return


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

We instantiate UAgents in this block “gemini_agent” We then register a startup event handler that logs when the agent is ready, ensuring visibility into its lifecycle. The startup handler logs the ready state of the agent, giving us visibility into its entire lifecycle. The ask_gemini synchronous wrapper wraps the GenAI call from the client to Gemini. “flash” model. The @ai_agent.on_message handling handler also deserializes the incoming question messages, calls ask_gemini and sends back an answer payload asynchronously to the sender. Take a look at the Notebook Here is a link to the article

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 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)
 # Shutdown with more grace
    asyncio.create_task(shutdown_loop())


async def shutdown_loop():
 Wait for asyncio.sleep(1) to allow time for cleaning
    loop = asyncio.get_event_loop()
    loop.stop()

The premise of the a “client_agent” This program, when it starts, asks the gemini_agent for the capital city of France. It then waits to hear the answer, prints out the response and gracefully closes the loop. Take a look at the Notebook Here is a link to the article

def run_agent(agent):
    agent.run()


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


    client_agent.run()


    p.join()

We define an agent.run function to help us run the Agent.()The gemini_agent is launched in the process using Python multiprocessing. It starts the gemini_agent, blocks until the round trip is complete, then launches the client_agent.

This tutorial focuses on UAgents. We now have a blueprint to create modular AI services which communicate using well-defined message and event schemas. UAgents makes it easy to manage agent lifecycles, handle incoming messages and send structured responses, without the need for boilerplate network code. You can now expand the UAgents configuration to add more complex conversation workflows, different message types and dynamic agent detection.

Take a look at the Notebook here. This research is the work of researchers. Also, feel free to follow us on Twitter Don’t forget about our 100k+ ML SubReddit Subscribe now our Newsletter.