Run time logic for excluding a subagent from a parent agent.

[ckl-axel]

Is your feature request related to a problem? Please describe.
We have an onboarding agent that is highly optimized for the initial interaction with new users. It would be great if we can have conditional inclusion of the sub agent so that outside of the initial interaction we know for certain it will not be transferred to. The condition can be a field in the session state of the user.

Describe alternatives you've considered
We are currently doing it via prompt engineering but a deterministic way to do this would be appreciated.

[hangfei]

@adk-bot

[adk-bot]

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

TLDR: You can achieve this deterministically by using either a before_agent_callback to conditionally skip the onboarding agent, or by creating a custom parent agent to handle the routing logic. Both methods can check a value in the session state to decide whether to run the onboarding sub-agent.

---

Hello @ckl-axel,

Thank you for your feature request. Having a deterministic way to control agent execution is a great point. ADK provides a couple of ways to achieve the conditional execution you're looking for based on the session state.

Using a before_agent_callback

This is likely the most direct way to solve your specific use case. You can attach a callback function to your onboarding sub-agent that runs before the agent itself. This function can inspect the session.state and, if a certain condition is met (e.g., the initial interaction is complete), it can prevent the agent from running.

Here’s a conceptual example based on the documentation [1]:

from adk.core import agents, callbacks, session, types

def check_if_onboarding_needed(context: callbacks.BeforeAgentContext) -> types.MaybeContent:
    # Check a flag in the session state
    if session.state.get("onboarding_complete"):
        # If onboarding is done, skip this agent by returning a Content object
        print("Skipping onboarding agent.")
        return types.Content(text="Skipping onboarding agent.")
    # Otherwise, allow the agent to run by returning None
    return None

# When creating your onboarding agent, attach the callback
onboarding_agent = agents.LLMAgent(
    # ... your agent config ...
)
onboarding_agent.add_callback(onboarding_agent.CallbackType.BEFORE_AGENT, check_if_onboarding_needed)

# In your main logic, after the first interaction, you would set the flag:
# session.state["onboarding_complete"] = True

Using a Custom Agent for Orchestration

For more complex conditional logic, you can create a custom parent agent that orchestrates its sub-agents. By implementing the _run_async_impl method, you have full control over the execution flow. [2]

Your custom agent could look something like this:

from adk.core import agents, session

class OnboardingOrchestrator(agents.BaseAgent):
    def __init__(self, onboarding_agent, main_agent):
        self._onboarding_agent = onboarding_agent
        self._main_agent = main_agent

    async def _run_async_impl(self, instructions: str):
        # Check the session state
        if not session.state.get("onboarding_complete"):
            # Run the onboarding agent
            response = await self._onboarding_agent.run_async(instructions)
            # Mark onboarding as complete
            session.state["onboarding_complete"] = True
            return response
        else:
            # Run the main agent
            return await self._main_agent.run_async(instructions)

Both of these approaches provide the deterministic control you are looking for by leveraging the session.state, just as you suggested. [3]

I hope this helps you build your desired agent structure!

[1] https://github.com/google/adk-docs/blob/main/examples/python/snippets/callbacks/before_agent_callback.py
[2] https://google.github.io/adk-docs/agents/custom-agents/
[3] https://google.github.io/adk-docs/agents/multi-agents/

[adk-bot]

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

TLDR: You can achieve this by implementing a custom agent. This agent will act as an orchestrator, checking a flag in the session state to decide whether to route the request to the onboarding sub-agent or other sub-agents.

---

Hello! That's a great question. You can definitely implement this kind of runtime logic in ADK.

While the sub_agents list for a parent agent is defined at initialization, you can create a custom agent that contains the logic to dynamically select which sub-agent to execute based on the current session state. This gives you the deterministic control you're looking for.

Here’s a conceptual example of how you could build an orchestrator agent that conditionally runs an onboarding agent:

from google.adk.agents import BaseAgent, LlmAgent
from google.adk.sessions import InvocationContext, Session

class OnboardingOrchestrator(BaseAgent):
    """A custom agent to orchestrate onboarding."""

    def __init__(self, onboarding_agent: BaseAgent, main_agent: BaseAgent, **kwargs):
        # Pass all potential sub-agents to the parent constructor
        # so the framework is aware of them for tracing and other purposes.
        super().__init__(sub_agents=[onboarding_agent, main_agent], **kwargs)
        self.onboarding_agent = onboarding_agent
        self.main_agent = main_agent

    async def _run_async_impl(self, ctx: InvocationContext):
        """
        Conditionally runs the onboarding agent or the main agent
        based on a flag in the session state.
        """
        # Check a flag in the session state.
        onboarding_complete = ctx.session.state.get("onboarding_complete", False)

        if not onboarding_complete:
            # If onboarding is not complete, run the onboarding agent.
            # The onboarding agent would be responsible for setting the
            # 'onboarding_complete' flag to True in the session state
            # when it's finished.
            async for event in self.onboarding_agent.run_async(ctx):
                yield event
        else:
            # Otherwise, run the main agent.
            async for event in self.main_agent.run_async(ctx):
                yield event

In this pattern:

1. OnboardingOrchestrator inherits from adk.agents.BaseAgent to become a custom agent. [1]
2. Its _run_async_impl method contains the core logic. It checks ctx.session.state for an onboarding_complete flag.
3. Based on the flag, it invokes either the onboarding_agent or the main_agent.
4. The onboarding_agent itself would be responsible for setting onboarding_complete: True in the session state upon completion of the onboarding flow.

This approach allows for complex and deterministic routing logic within your agent hierarchy.

For more details, you can refer to the official documentation on creating custom agents. [1]

---

[1] https://google.github.io/adk-docs/agents/custom-agents/