In our last lesson, we created the complete blueprint for our "Lyricist" agent by defining its skills and its public-facing Agent Card. However, this blueprint is just a plan. The agent doesn't have an engine to do the actual work, and it isn't connected to the internet yet. In this lesson, we will solve both of those problems.

First, we will build the agent's "engine" using a class called the AgentExecutor. Then, we will launch a simple web server, making our agent live and ready to receive requests from other agents.

Step 1: The Agent Executor

The Agent Executor is the core logic that connects the A2A server to our agent's brain. When a request comes in, the server passes it to the executor. The executor's job is to understand the request, ask our agent to perform the work, and then send the result back.

To do this, we create a class that inherits from the AgentExecutor base class provided by the SDK. We then need to implement a key method called execute.

The execute method gives us two important tools:

context: This object contains all the information about the incoming request, such as the user's message.
event_queue: This is the channel we use to send responses back to the user.

Let's create an executor for our Lyricist agent. This executor will take the incoming text, pass it to the lyricist_agent we created in Lesson 3, and enqueue the result.

Coding agents often struggle to generate code using new libraries like a2a-sdk and google-adk. We recommend to use our code examples or give it as a context to your coding agent for the best results.

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore
from a2a.utils import new_agent_text_message
from a2a.types import (
    Task,
    UnsupportedOperationError
)

# This is our agent's "engine"
class LyricistAgentExecutor(AgentExecutor):
    def __init__(self, agent):
        self.agent = agent

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # Get the incoming text from the user's message
        prompt = context.message.text
        
        # Ask our agent to generate the lyrics
        result = await self.agent.chat(prompt)

        # Send the result back as a new text message
        await event_queue.enqueue_event(new_agent_text_message(result.text))

    async def cancel(
        self, request: RequestContext, event_queue: EventQueue
    ) -> Task | None:
        raise ServerError(error=UnsupportedOperationError())

Step 2: Setting Up and Starting the Server

Now that we have our Agent Card and the Agent Executor, we can start our server. The a2a-sdk makes this very easy with the A2AStarletteApplication class, which uses the popular Uvicorn server to run.

We need to do three things:

Create a Request Handler: The DefaultRequestHandler connects our LyricistAgentExecutor to the server's routing logic.
Create the Server Application: The A2AStarletteApplication takes our agent's public Agent Card and the request handler. The server will automatically make the Agent Card available at the /.well-known/agent.json URL.
Run the Server: We use the uvicorn.run() function to start the web server.

Step 3: Putting It All Together in `main.py`

Let's combine everything we have built in previous lessons into a single, complete main.py file. Add the following code at the end of the file or ask your coding agent to do it for you.

# 5. The Server Setup
if __name__ == '__main__':
    # Connect the engine to the request handler
    request_handler = DefaultRequestHandler(
        agent_executor=LyricistAgentExecutor(agent=lyricist_agent),
        task_store=InMemoryTaskStore(),
    )

    # Create the server application with our agent's card and handler
    server = A2AStarletteApplication(
        agent_card=lyricist_agent_card,
        http_handler=request_handler,
    )

    # Start the server!
    print("Starting Lyricist Agent server on http://localhost:9999")
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

Full main.py code example


import uvicorn
from google.adk import Agent
from a2a.types import AgentSkill, AgentCard, AgentCapabilities
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore
from a2a.utils import new_agent_text_message
from a2a.types import (
    Task,
    UnsupportedOperationError
)

# 1. The Agent's "Brain"
lyricist_agent = Agent(
    name='lyricist_agent',
    description="A creative AI agent that specializes in writing song lyrics.",
    instruction="You are a lyricist agent. You need to write song lyrics based on the prompt."
)

# 2. The Agent's Skill
generate_lyrics_skill = AgentSkill(
    id='generate_lyrics',
    name='Generate Song Lyrics',
    description='Creates creative song lyrics based on a given theme.',
    tags=['creative', 'music', 'writing', 'lyrics'],
    examples=['write a song about exploring the ocean']
)

# 3. The Agent's Public "Business Card"
lyricist_agent_card = AgentCard(
    name=lyricist_agent.name,
    description=lyricist_agent.description,
    url='http://localhost:9999',
    version='1.0.0',
    defaultInputModes=['text/plain'],
    defaultOutputModes=['text/plain'],
    capabilities=AgentCapabilities(streaming=False),
    skills=[generate_lyrics_skill]
)

# This is our agent's "engine"
class LyricistAgentExecutor(AgentExecutor):
    def __init__(self, agent):
        self.agent = agent

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # Get the incoming text from the user's message
        prompt = context.message.text
        
        # Ask our agent to generate the lyrics
        result = await self.agent.chat(prompt)

        # Send the result back as a new text message
        await event_queue.enqueue_event(new_agent_text_message(result.text))

    async def cancel(
        self, request: RequestContext, event_queue: EventQueue
    ) -> Task | None:
        raise ServerError(error=UnsupportedOperationError())

# 5. The Server Setup
if __name__ == '__main__':
    # Connect the engine to the request handler
    request_handler = DefaultRequestHandler(
        agent_executor=LyricistAgentExecutor(agent=lyricist_agent),
        task_store=InMemoryTaskStore(),
    )

    # Create the server application with our agent's card and handler
    server = A2AStarletteApplication(
        agent_card=lyricist_agent_card,
        http_handler=request_handler,
    )

    # Start the server!
    print("Starting Lyricist Agent server on http://localhost:9999")
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

Step 4: Run the Server

Now, it's time to start our agent! Open your terminal, make sure your virtual environment is activated, and run the main file.

python main.py

You should see output in your terminal indicating that the Uvicorn server is running. This means your agent is now live and listening for requests on your local machine!

Outcome of This Lesson and What's Next

Congratulations! You have successfully taken your agent from a plan to a live, running application. We have built the agent's core logic with an AgentExecutor and launched a web server to make it accessible.

Our "Lyricist" agent is ready, but it's currently working in isolation. In our next lesson, we will learn how another agent—our "Composer"—can discover and communicate with the Lyricist to create a complete, collaborative piece of work.

Yay! Project completed!

Rate your experience