Update workflow_concurrent_research_writer sample to the latest ADK Workflow API

pyproject.toml

@@ -21,6 +21,11 @@ build-backend = "hatchling.build"
 url = "https://pypi.org/simple/"
 default = true
 
+[tool.uv.workspace]
+members = [
+    "python/agents/workflow-concurrent_research_writer",
+]
+
 [tool.hatch.build.targets.wheel]
 packages = ["tools"]
 

python/agents/workflow_concurrent_research_writer/.python-version

@@ -0,0 +1 @@
+3.12

python/agents/workflow_concurrent_research_writer/__init__.py

@@ -0,0 +1 @@
+from . import agent

python/agents/workflow-concurrent_research_writer/agent.py → python/agents/workflow_concurrent_research_writer/agent.py

@@ -2,12 +2,11 @@
 from functools import partial
 
 # from uuid import uuid4
-from google.adk.agents.workflow.base_node import START
+from google.adk.workflow import START
 
 # from google.adk.agents.workflow.join_node import JoinNode
-from google.adk.agents.workflow.function_node import FunctionNode
-from google.adk.agents.workflow.parallel_worker import ParallelWorker
-from google.adk.agents.workflow.workflow_agent import WorkflowAgent
+from google.adk.workflow import FunctionNode
+from google.adk.workflow import Workflow
 
 # from google.adk.agents.run_config import RunConfig
 # from google.adk.sessions.in_memory_session_service import InMemorySessionService
@@ -18,7 +17,7 @@
 from .agent_nodes.publishing import generate_blog_post_agent
 from .agent_nodes.research import (
     distill_agent,
-    research_worker_agent,
+    parallel_research,
 )
 from .function_nodes.publishing import (
     post_node,
@@ -33,13 +32,13 @@
 # Research Workflow: A simple, linear chain. The `research_worker_agent`
 # is marked with `parallel_worker=True` so the framework will automatically
 # handle fanning out for each query and fanning in the results.
-research_workflow = WorkflowAgent(
+research_workflow = Workflow(
     name="research_workflow",
     edges=[
         (
             START,
             start_node,
-            ParallelWorker(research_worker_agent),
+            parallel_research,
             distill_agent,
             save_node,
         ),
@@ -48,49 +47,70 @@
 
 # Blog Workflow
 # Nodes for posting the main article
-post_to_x = FunctionNode(partial(post_node, "X"), name="Post to X")
+post_to_x = FunctionNode(func=partial(post_node, "X"), name="Post_to_X")
 post_to_linkedin = FunctionNode(
-    partial(post_node, "LINKEDIN"), name="Post to LinkedIn"
+    func=partial(post_node, "LINKEDIN"), name="Post_to_LinkedIn"
 )
 post_to_medium = FunctionNode(
-    partial(post_node, "MEDIUM"), name="Post to Medium"
+    func=partial(post_node, "MEDIUM"), name="Post_to_Medium"
 )
 
 # Nodes for posting shoutouts
-shoutout_to_x = FunctionNode(partial(shoutout_node, "X"), name="Shoutout to X")
+shoutout_to_x = FunctionNode(func=partial(shoutout_node, "X"), name="Shoutout_to_X")
 shoutout_to_linkedin = FunctionNode(
-    partial(shoutout_node, "LINKEDIN"), name="Shoutout to LinkedIn"
+    func=partial(shoutout_node, "LINKEDIN"), name="Shoutout_to_LinkedIn"
 )
 shoutout_to_medium = FunctionNode(
-    partial(shoutout_node, "MEDIUM"), name="Shoutout to Medium"
+    func=partial(shoutout_node, "MEDIUM"), name="Shoutout_to_Medium"
 )
 shoutout_to_reddit = FunctionNode(
-    partial(shoutout_node, "REDDIT"), name="Shoutout to Reddit"
+    func=partial(shoutout_node, "REDDIT"), name="Shoutout_to_Reddit"
 )
 
-blog_workflow = WorkflowAgent(
+blog_workflow = Workflow(
     name="blog_workflow",
     edges=[
         # 1. Start, write blog, then route by length
         (START, start_blog, generate_blog_post_agent, route_changer),
         # 2. Post to the primary platform based on the route from route_changer
-        (route_changer, post_to_x, "X"),
-        (route_changer, post_to_linkedin, "LINKEDIN"),
-        (route_changer, post_to_medium, "MEDIUM"),
+        (route_changer,
+            {
+                "X": post_to_x,
+                "LINKEDIN": post_to_linkedin,
+                "MEDIUM": post_to_medium,
+            },
+        ),
         # 3. From each primary post, trigger shoutouts based on the new objective rules.
         # If posted to X -> Shoutout to LinkedIn and Reddit
-        (post_to_x, shoutout_to_linkedin, "SHOUTOUT_LINKEDIN"),
-        (post_to_x, shoutout_to_reddit, "SHOUTOUT_REDDIT"),
+        (
+            post_to_x, 
+            {
+                "SHOUTOUT_LINKEDIN":shoutout_to_linkedin,
+                "SHOUTOUT_REDDIT":shoutout_to_reddit,
+            },
+        ),
+
         # If posted to LinkedIn -> Shoutout to X and Reddit
-        (post_to_linkedin, shoutout_to_x, "SHOUTOUT_X"),
-        (post_to_linkedin, shoutout_to_reddit, "SHOUTOUT_REDDIT"),
+        (
+            post_to_linkedin,
+            {
+                "SHOUTOUT_X":shoutout_to_x,
+                "SHOUTOUT_REDDIT":shoutout_to_reddit,
+            },
+        ),
         # If posted to Medium -> Shoutout to X and LinkedIn
-        (post_to_medium, shoutout_to_x, "SHOUTOUT_X"),
-        (post_to_medium, shoutout_to_linkedin, "SHOUTOUT_LINKEDIN"),
+        (
+            post_to_medium,
+            {
+                "SHOUTOUT_X":shoutout_to_x,
+                "SHOUTOUT_LINKEDIN": shoutout_to_linkedin
+            }
+
+        ),        
     ],
 )
 
-root_agent = WorkflowAgent(
+root_agent = Workflow(
     name="root_agent",
     description="""
         Main workflow contucting the research and pubication phases of blog 

python/agents/workflow-concurrent_research_writer/agent_nodes/publishing.py → python/agents/workflow_concurrent_research_writer/agent_nodes/publishing.py

@@ -1,5 +1,5 @@
 from google.adk.agents.llm_agent import LlmAgent
-from src.prompts import GENERATE_BLOG_POST_PROMPT
+from ..prompts import GENERATE_BLOG_POST_PROMPT
 
 generate_blog_post_agent = LlmAgent(
     name="generate_blog_post_agent",

python/agents/workflow_concurrent_research_writer/agent_nodes/research.py

@@ -0,0 +1,120 @@
+from google.adk.agents.llm_agent import LlmAgent
+from ..prompts import (
+    JOIN_AND_DISTILL_PROMPT,
+)
+import asyncio
+from ..tools import execute_search
+from google.adk.workflow import FunctionNode
+
+import uuid
+
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+from google.genai.types import Content, Part
+
+APP_NAME = "parallel_research"
+
+# The LlmAgent that will be run in parallel for each platform.
+_research_worker_llm_agent = LlmAgent(
+    # The name for the inner agent is not strictly necessary for the
+    # workflow but can be useful for debugging.
+    name="research_worker_llm_agent",
+    model="gemini-2.5-flash",
+    instruction="""Your sole task is to research the topic '{topic}' on a specific platform.
+The platform you MUST use is provided as your input. Your entire input is the name of the platform.
+DO NOT ask for the platform. Use the input you are given.
+Execute a search on that platform for the topic and summarize the results.""",
+    tools=[execute_search],
+)
+
+
+from pydantic import BaseModel
+
+class DistillInput(BaseModel):
+    research: str
+
+# The Synthesizer joins the results from the parallel workers into a final report.
+distill_agent = LlmAgent(
+    name="join_and_distill_agent",
+    model="gemini-2.5-flash",
+    input_schema=DistillInput,
+    instruction=JOIN_AND_DISTILL_PROMPT,
+    output_schema=str
+)
+
+session_service = InMemorySessionService()
+
+research_runner = Runner(
+    app_name=APP_NAME,
+    agent=_research_worker_llm_agent,
+    session_service=session_service,
+)
+
+async def run_research_agent(topic, platform) -> str:
+    """
+    Executes _research_worker_llm_agent for a single platform.
+
+    Returns the final generated text.
+    """
+
+    user_id = "parallel-worker"
+
+    session_id = str(uuid.uuid4())
+
+    # Every parallel invocation gets its own session
+    await session_service.create_session(
+        app_name=APP_NAME,
+        user_id=user_id,
+        session_id=session_id,
+        state={
+            "topic": topic,
+        },
+    )
+
+    user_message = Content(
+        role="user",
+        parts=[
+            Part.from_text(text=platform),
+        ],
+    )
+
+    print(f"user_message -- {user_message}")
+
+    final_text = ""
+
+    async for event in research_runner.run_async(
+        user_id=user_id,
+        session_id=session_id,
+        new_message=user_message,
+    ):
+        if event.is_final_response():
+            if (
+                event.content
+                and event.content.parts
+                and event.content.parts[0].text
+            ):
+                final_text = event.content.parts[0].text
+    print(f"final text -- \n\n {final_text}")
+    return final_text
+
+from google.genai.types import Content, Part
+
+
+async def parallel_research(
+    node_input: list[str],
+    topic:str
+) -> DistillInput:
+    platforms_to_research = node_input
+    print(f"topic : {topic} ")
+    reports = await asyncio.gather(
+        *[
+            run_research_agent(topic, platform)
+            for platform in platforms_to_research
+        ]
+    )
+
+    combined = "\n\n---\n\n".join(reports)
+    print(f"Combined -- {combined}")
+    return DistillInput(
+        research=combined
+    )

python/agents/workflow-concurrent_research_writer/function_nodes/publishing.py → python/agents/workflow_concurrent_research_writer/function_nodes/publishing.py

@@ -1,9 +1,10 @@
 from collections.abc import AsyncGenerator
 
-from google.adk.agents.workflow.events.event import Event
-from google.adk.agents.workflow.function_node import FunctionNode
+from google.adk import Event
+from google.adk.agents import InvocationContext
+from google.adk.workflow import FunctionNode
 from google.genai.types import Content
-from src.tools import post_to_platform
+from ..tools import post_to_platform
 
 # --- Workflow 2 Nodes: Blog and Publish ---
 
@@ -12,10 +13,12 @@
 
 
 async def start_blog_node(
-    node_input: Content,
+    ctx: InvocationContext,
 ) -> AsyncGenerator[Event | str, None]:
     """Entry node for the blog workflow, takes user thesis from Content object."""
-    thesis = str(node_input.parts[0].text if node_input.parts else "")
+    print("START_WORKFLOW 2  K: Blog post with thesis: ")
+    thesis = ctx.state["research_report"]
+    # thesis = str(node_input.parts[0].text if node_input.parts else "")
     print(f"START_WORKFLOW 2: Blog post with thesis: '{thesis}'")
     # The thesis is passed as the node_input to the next agent in the chain.
     yield thesis
@@ -65,11 +68,12 @@ async def post_node(
 
 async def shoutout_node(platform: str, node_input: str) -> None:
     """Posts a shoutout to a given platform."""
+    print(f"Shoutout to the following platform {platform} ")
     blog_post_preview = node_input[:40].strip()
     shoutout_msg = f"Check out my new article! '{blog_post_preview}...'"
     await post_to_platform(platform, shoutout_msg, shoutout=True)
 
 
 # Function node Wrappers
-start_blog = FunctionNode(start_blog_node, name="Start Blog Writing")
-route_changer = FunctionNode(length_router_node, name="PathFinder")
+start_blog = FunctionNode(func=start_blog_node, name="Start_Blog_Writing")
+route_changer = FunctionNode(func=length_router_node, name="PathFinder")