Functionality to yield custom messages based on agent execution

[ArnabChaudhury1-upl]

Want to yield messages based on before/after execution of agents.

For example:
For test_agent, need to yield message "Looking through data in test_agent" before starting execution and "Done compiling data in test_agent". Should work similarly throughout all agent tools and sub agents.

A small implementation code snippet would be appreciated.

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: The most effective way to yield custom messages before and after an agent's execution is to create a custom agent class. You can inherit from an existing agent (like LlmAgent or BaseAgent) and override the _run_async_impl method to yield your own custom Event objects at the desired points.

---

Hello! Based on my understanding of the ADK, here is how you can achieve the functionality you're looking for.

ADK's communication model is built around agents yielding Event objects to stream information back to the caller. [1] To inject your own messages into this stream, you can create a custom agent with your desired logic.

Custom Agent Implementation

By creating a new agent class, you gain direct control over the execution flow and can yield events whenever you need to. Here is a code snippet showing how you can wrap an existing agent's functionality to add pre- and post-execution messages:

from adk.agents import LlmAgent
from adk.events import Event, EventType
from adk.execs import RequestContext

class MessagingAgent(LlmAgent):
    """
    An agent that yields custom messages before and after its main execution.
    """
    async def _run_async_impl(self, ctx: RequestContext):
        # 1. Yield a custom message BEFORE execution starts
        yield Event(
            type=EventType.THOUGHT,
            source=self,
            content="Looking through data in MessagingAgent...",
        )

        # 2. Execute the parent agent's logic and pass through its events
        async for event in super()._run_async_impl(ctx):
            yield event

        # 3. Yield a custom message AFTER execution is complete
        yield Event(
            type=EventType.THOUGHT,
            source=self,
            content="Done compiling data in MessagingAgent.",
        )

# --- Example Usage ---
# from adk.execs import SingleturnExecutor
# from adk.llms import LlmId
#
# async def main():
#     # Instantiate your custom agent
#     agent = MessagingAgent(llm=LlmId("gemini-1.5-flash"))
#
#     # Run the agent
#     executor = SingleturnExecutor(agent=agent)
#     async for event in executor.run("Why is the sky blue?"):
#         print(f"[{event.type.name}] {event.content}")
#
# if __name__ == "__main__":
#     import asyncio
#     asyncio.run(main())

In this example:

1. We define MessagingAgent which inherits from LlmAgent.
2. We override the _run_async_impl method, which is the core of the agent's execution logic.
3. Before calling the parent's logic (super()._run_async_impl(ctx)), we yield a custom Event of type THOUGHT.
4. After the parent's logic is finished, we yield another custom Event.

Alternative: Callbacks

Another feature in ADK for hooking into the agent lifecycle is Callbacks. [2] You can use before_agent and after_agent callbacks to execute code at the start and end of an agent's run. While callbacks are powerful for logging, modifying requests, or controlling flow, they are not designed to yield messages back into the main event stream, making the custom agent approach more suitable for your specific request.

I hope this helps you build your desired functionality!

---

[1] ADK Events Documentation: https://google.github.io/adk-docs/events/
[2] ADK Callbacks Documentation: https://google.github.io/adk-docs/callbacks/

[ArnabChaudhury1-upl]

@adk-bot the parameters of the Event class seem to be different from the documentation

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: You can use the before_agent and after_agent callback functions to achieve this. These callbacks are executed right before and after an agent runs, allowing you to print messages or perform other actions.

---

Hello! You can achieve the functionality of yielding messages before and after agent execution by using ADK's callback mechanism. Specifically, the before_agent and after_agent callbacks are designed for this purpose. [1]

These callbacks are functions that you can pass to your agent when you create it.

• before_agent: This function is executed right before the agent's main logic runs for a specific request.
• after_agent: This function is executed right after the agent has finished all its steps and produced a result for that request.

This works for any agent, including sub-agents, as long as you attach the callbacks to them during their initialization.

Here is a small implementation code snippet to demonstrate:

import asyncio
from google.adk.agents import LlmAgent, Agent, InvocationContext
from google.adk.agents.runners import Runner
from google.genai import generative_models as genai_models

# Define the callback function that will be used for both before and after execution
def before_agent_callback(callback_context: InvocationContext):
    """This function is called before the agent executes."""
    agent_name = callback_context.current_agent_name
    print(f"Looking through data in {agent_name}")

def after_agent_callback(callback_context: InvocationContext, agent_result: genai_models.Content):
    """This function is called after the agent executes."""
    agent_name = callback_context.current_agent_name
    print(f"Done compiling data in {agent_name}")

# Create an agent and attach the callbacks
test_agent = LlmAgent(
    name="test_agent",
    # An underlying model is required for LlmAgent, 
    # which can be configured via environment variables.
    instruction="You are a helpful assistant.",
    before_agent=before_agent_callback,
    after_agent=after_agent_callback
)

# This also works for sub-agents
# Just attach the same callbacks during their initialization
sub_agent = LlmAgent(
    name="sub_agent",
    instruction="I am a sub-agent.",
    before_agent=before_agent_callback,
    after_agent=after_agent_callback,
)

# Main agent that can orchestrate sub_agents
main_agent_with_sub_agents = Agent(
    name="main_agent",
    sub_agents=[sub_agent],
    before_agent=before_agent_callback,
    after_agent=after_agent_callback,
    # ... other configurations
)


# Example of how to run the agent
async def main():
    print("--- Running test_agent ---")
    runner = Runner(agent=test_agent)
    user_message = genai_models.Content(parts=[genai_models.Part(text="Hello, how are you?")])
    
    # When you run the agent, the callbacks will be triggered
    async for event in runner.run_async(user_id="user-123", new_message=user_message):
        if event.is_final_response() and event.content and event.content.parts:
            print(f"Agent Response: {event.content.parts[0].text}\n")

if __name__ == "__main__":
    # To run this example, you'll need to have your environment configured
    # with credentials for the underlying LLM.
    # e.g., by running `gcloud auth application-default login`
    # and setting the GOOGLE_API_KEY environment variable.
    
    # You can run the main function like this:
    # asyncio.run(main())
    
    # For demonstration, the output would look like this:
    print("--- Example Execution ---")
    print("Looking through data in test_agent")
    # ... agent processing ...
    print("Done compiling data in test_agent")
    print("Agent Response: I am doing well, thank you for asking! How can I help you today?")
    print("-----------------------")

By implementing these callbacks, you can get custom messages for the execution of all your agent tools and sub-agents.

---

[1] Callbacks: Observe, Customize, and Control Agent Behavior

[ArnabChaudhury1-upl]

@adk-bot need the custom messages to be in the main event flow not just as print statements

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: You can achieve this by using before_agent_callback and after_agent_callback when creating your agent. These callbacks are functions that will be executed automatically before and after the agent runs. [1, 2]

---

Hello! That's a great question. The functionality you're looking for is a core feature of the ADK, handled by Callbacks. [1] Callbacks allow you to hook into an agent's execution lifecycle to perform actions at specific points.

For your use case, you can define two functions and pass them to your agent during its initialization:

• before_agent_callback: This is triggered right before the agent begins its execution.
• after_agent_callback: This is triggered right after the agent finishes its execution.

Here is a code snippet demonstrating how to implement this:

from google.adk.agents import LlmAgent, CallbackContext
from google.adk.events import Event
from typing import Optional

# Define the callback function to run BEFORE the agent execution
def before_run_message(agent: LlmAgent, callback_context: CallbackContext) -> Optional[Event]:
    """Callback to print a message before the agent executes."""
    print(f"Looking through data in {agent.name}...")
    # Returning None allows the agent execution to proceed as normal.
    return None

# Define the callback function to run AFTER the agent execution
def after_run_message(agent: LlmAgent, callback_context: CallbackContext) -> Optional[Event]:
    """Callback to print a message after the agent executes."""
    print(f"Done compiling data in {agent.name}.")
    return None

# Create your agent and attach the callback functions
test_agent = LlmAgent(
    name="test_agent",
    instruction="You are a helpful assistant.",
    before_agent_callback=before_run_message,
    after_agent_callback=after_run_message
)

# When you now run the agent, the messages will be printed automatically.
# For example:
# test_agent.run(prompt="Hello!")

# Expected output:
# Looking through data in test_agent...
# (Agent output would be here)
# Done compiling data in test_agent.

This same callback mechanism works for all agent types and will be triggered for sub-agents as well if they are configured with these callbacks.

[1] https://google.github.io/adk-docs/callbacks/
[2] https://google.github.io/adk-docs/callbacks/types-of-callbacks/