Google ADK

agents-google-adk/
├── .env                 # GOOGLE_API_KEY=your_key_here
├── requirements.txt
└── blogger/
    ├── __init__.py
    ├── agent.py         # All agents, routing functions, root_agent (Workflow)
    └── main.py          # Runner, session service, async entry point

google-adk==2.2.0
python-dotenv>=1.0.0

from datetime import date

from dotenv import load_dotenv
from google.adk.agents import Agent, Context
from google.adk.workflow import Edge, START, Workflow, node

load_dotenv()
MODEL = "gemini-flash-latest"

blog_planner = Agent(
    name="BlogPlanner",
    description="Creates a practical, skimmable outline in Markdown",
    model=MODEL,
    output_key="blog_outline",
    instruction="""
As a technical content strategist, produce a clear Markdown outline with:
- Title
- Short intro
- 2 main sections.
- Conclusion

If `codebase_context` exists in state, weave in specific sections/snippets.
Return only the outline in Markdown.
""",
)

outline_validator = Agent(
    name="OutlineValidator",
    description="Validates the outline.",
    model=MODEL,
    instruction="""
Check the outline (`blog_outline`) in state. If it has a title, intro,
2 sections and a conclusion, respond "OK", otherwise return "RETRY" and
list missing pieces.""",
    output_key="validation_result",
)

def _route_outline(ctx: Context) -> str:
    result = ctx.state.get("validation_result", "")
    tries = ctx.state.get("planner_tries", 0) + 1
    ctx.state["planner_tries"] = tries
    return "OK" if result.strip().upper().startswith("OK") or tries >= 3 else "RETRY"

route_outline = node(_route_outline, name="RouteOutline")

resilient_planner = Workflow(
    name="ResilientPlanner",
    description="Generates and validates a blog outline, retrying up to 3 times",
    edges=[
        (START, blog_planner, outline_validator, route_outline),
        Edge(from_node=route_outline, to_node=blog_planner, route="RETRY"),
    ],
)

# The writer stage mirrors the planner stage exactly: an agent that
# produces, a validator that judges, and a routing function that loops.

def _route_post(ctx: Context) -> str:
    result = ctx.state.get("validation_result", "")
    tries = ctx.state.get("writer_tries", 0) + 1
    ctx.state["writer_tries"] = tries
    return "OK" if result.strip().upper().startswith("OK") or tries >= 3 else "RETRY"

route_post = node(_route_post, name="RoutePost")

blog_writer = Agent(
    name="BlogWriter",
    description="Blog Writer Agent",
    instruction="""
Write a complete Markdown article from the outline in `blog_outline`
Guidelines:
- Audience: Software Engineers
- Explain both the `how` and `why`
- Include concise code snippets when helpful
- Output only the final article in Markdown (no fence around the whole post)""",
    model=MODEL,
    output_key="blog_post",
)

post_validator = Agent(
    name="BlogPostValidator",
    description="Validates the final post.",
    model=MODEL,
    instruction="""
Check the final post (`blog_post`) in state. If it has a title, intro,
clear sections matching the outline, conclusion and is technically accurate
respond "OK", otherwise return "RETRY" with the specific fixes.""",
    output_key="validation_result",
)

resilient_writer = Workflow(
    name="ResilientWriter",
    description="Generates and validates a blog post, retrying up to 3 times",
    edges=[
        (START, blog_writer, post_validator, route_post),
        Edge(from_node=route_post, to_node=blog_writer, route="RETRY"),
    ],
)

final_agent = Agent(
    name="FinalAgent",
    description="Generates alternative titles and tweet-length hooks",
    model=MODEL,
    instruction=f"""
Based on the blog post in `blog_post` state, provide:
- 2 alternative titles
- 2 tweet-length hooks

Date: {date.today().isoformat()}
""",
)

root_agent = Workflow(
    name="Blogger",
    description="Plans and writes a complete blog post end-to-end.",
    edges=[
        (START, resilient_planner, resilient_writer, final_agent),
    ],
)

import asyncio
import sys

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from .agent import root_agent

async def run(topic: str) -> None:
    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name="blogger", user_id="local", session_id="run-1"
    )
    runner = Runner(
        node=root_agent,
        app_name="blogger",
        session_service=session_service,
    )
    async for event in runner.run_async(
        user_id="local",
        session_id="run-1",
        new_message=types.Content(
            role="user",
            parts=[types.Part(text=topic)],
        ),
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(part.text, end="", flush=True)
    print()


if __name__ == "__main__":
    TOPIC = " ".join(sys.argv[1:]) or "Introduction to Google ADK multi-agent systems"
    asyncio.run(run(TOPIC))