DEV Community: YAMASAKI Masahide

🚀 Smarter Google ADK Prompts: Inject State and Artifact Data Dynamically Placeholders

YAMASAKI Masahide — Tue, 29 Apr 2025 12:25:21 +0000

If you're building AI agents using Google's Agent Development Kit (ADK), you might have asked:

"How can I make my agent prompts context-aware without hardcoding a ton of data?"

Good news — while exploring how LlmAgent handles session data and artifacts, we discovered a built-in (and kind of undocumented 👀) feature: placeholder substitution in the instruction field.

Let’s explore how this feature works and how you can use it to dynamically inject values from session.state and ArtifactService into your prompts — cleanly and automatically.

🎯 The Problem: Making Prompts Smarter

Agents often need to reference:

A user's previous input
Output from another agent/tool
A document or file uploaded earlier

ADK gives us tools like session.state and ArtifactService to manage that data.

But manually inserting all that into the prompt? 😵 That’s messy, hard to maintain, and likely to hit LLM context size limits.

Wouldn’t it be nice if your instruction could just say something like {user_name} or {artifact.summary.txt} — and the framework filled in the blanks?

🧩 The Solution: Built-In Placeholder Substitution

Turns out, you can do exactly that.

After diving into ADK's internals (especially _populate_values in instructions.py), we confirmed that ADK supports templated instructions using {} syntax. These placeholders are evaluated before sending the prompt to the LLM.

🧠 Supported Placeholder Types

Here’s what you can use:

1. Session State

{key} or {state.key} → looks up session.state["key"]
{app:key} → gets value from app-scoped state
{user:key} → gets value from user-scoped state
{key?} or {state.key?} → optional placeholder (returns empty string if key not found)

All values are converted to string with str() before being inserted.

2. Artifact Content

{artifact.filename} → inserts the content of the given artifact file (if available)

🧪 Confirmed in Tests

We found unit tests in test_instructions.py that confirm this behavior.

Tests like test_build_system_instruction verify that placeholders like {customerId}, {app:key}, {user:key}, and {artifact.filename} are correctly replaced with session or artifact data during instruction building.

✅ The placeholder feature is clearly supported both by the ADK source code and validated through unit tests!

🛠️ Example 1: Using State in Prompts

Let’s say a generator_agent writes a draft, and a reviewer_agent checks it.

from google.adk.agents import Agent

# Generator saves its output into session.state['draft_text']
generator = Agent(
    name="generator_agent",
    model="gemini-2.0-flash",
    instruction="Write a short paragraph about subject X.",
    output_key="draft_text"
)

# Reviewer reads the state and uses it in the prompt
reviewer = Agent(
    name="reviewer_agent",
    model="gemini-2.0-flash",
    instruction="Please review the following draft: {draft_text}. Check for factual accuracy and provide feedback."
)

No need to manually pass variables around — ADK handles it for you 🙌

📁 Example 2: Using Artifacts in Prompts

from google.adk.agents import Agent

summarizer = Agent(
    name="summarizer_agent",
    model="gemini-2.0-flash",
    instruction="Summarize the document provided in the artifact named 'meeting_notes.txt': {artifact.meeting_notes.txt}"
)

This is a clean way to reference file contents without bloating your code or prompt logic.

🔄 Alternative: Build Instructions in Code (If You Must)

If you want more control, you can still build prompts manually using callbacks:

def update_instruction_before_call(callback_context: CallbackContext, llm_request: LlmRequest):
    draft = callback_context.state.get('draft_text', '')
    llm_request.config.system_instruction = f"Please review the following draft: {draft}. Check for factual accuracy."
    return None

reviewer = Agent(
    name="reviewer_agent",
    model="gemini-2.0-flash",
    instruction=None,
    before_model_callback=update_instruction_before_call
)

It works — but for simple cases, placeholder substitution is much cleaner ✨

🧭 Final Thoughts

The placeholder feature ({state.key}, {artifact.filename}) lets you write smarter, more flexible agent instructions without string hacking or bloated callbacks.

💡 Important Note:

While this behavior is clearly present in the source code and unit tests, it is not emphasized in the official documentation.

In fact, the official example in the docs here shows state referencing using manual Python string interpolation instead of placeholders — which slightly contradicts the built-in behavior we observed.

So, while it works beautifully today, this feature may be considered an internal implementation detail, and future ADK versions could potentially change or formalize it.

Always check the latest ADK release notes when upgrading! 🚀

Use it wisely, and you’ll make your ADK agents a whole lot more powerful.

👍 Found this helpful? Let’s connect in the comments — and share if you discover any other hidden ADK tricks!

🚀 Building Dynamic Parallel Workflows in Google ADK

YAMASAKI Masahide — Sun, 27 Apr 2025 04:43:00 +0000

In real-world systems, you rarely know the exact number of tasks or agents you'll need ahead of time.

How can we dynamically assign and parallelize tasks based on runtime conditions?

In Google's Agent Development Kit (ADK),

the answer is simple but powerful:

Dynamically create a ParallelAgent inside a Custom Agent, and use session.state as your shared communication bus.

In this post, we'll cover:

✅ Why dynamic parallelism matters

✅ A minimal working example

✅ How to run and verify it

✅ Common pitfalls and best practices

✅ Extension ideas for scaling

🤔 Why Static Parallel Agents Aren't Enough

ADK provides several workflow primitives:

SequentialAgent (sequential execution) ⏩
ParallelAgent (parallel execution) 🔀
LoopAgent (repetitive execution) 🔄

ParallelAgent, in particular, is extremely lightweight —

it simply concurrently executes the given sub-agents.

Perfect for speeding up I/O-bound operations.

However, it has a key limitation:

✅ The list of child agents must be predefined at construction time.

That means:

You can't dynamically change how many workers run
You can't adapt to different task types at runtime

👉 To solve this, we build a fresh ParallelAgent dynamically inside a Custom Agent every time.

🗒️ Example: A 60-line Dynamic Fanout Pattern

Let's start with a minimal working example:

import random, secrets
from typing import ClassVar, List
from google.adk.events import Event, EventActions
from google.adk.agents import BaseAgent, ParallelAgent, SequentialAgent
from google.genai import types

class Worker(BaseAgent):
    """Simple worker that calculates n²."""
    def __init__(self, *, name: str, run_id: str):
        super().__init__(name=name); self._run_id = run_id
    async def _run_async_impl(self, ctx):
        n = ctx.session.state.get(f"task:{self._run_id}:{self.name}", 0)
        result = n * n
        yield Event(
            author=self.name,
            content=types.Content(role=self.name,
                                  parts=[types.Part(text=f"{n}² = {result}")]),
            actions=EventActions(
                state_delta={f"result:{self._run_id}:{self.name}": result})
        )

class PlannerAndRunner(BaseAgent):
    """Distributes tasks and dynamically creates a ParallelAgent."""
    POOL: ClassVar[List[str]] = ["w0", "w1", "w2"]
    async def _run_async_impl(self, ctx):
        run_id = secrets.token_hex(2)
        picked = random.sample(self.POOL,
                               k=random.randint(1, len(self.POOL)))
        task_delta = {f"task:{run_id}:{name}": random.randint(1, 9)
                      for name in picked}
        yield Event(
            author=self.name,
            content=types.Content(role=self.name,
                   parts=[types.Part(text=f"Run {run_id} tasks {task_delta}")]),
            actions=EventActions(state_delta={"current_run": run_id, **task_delta})
        )
        parallel = ParallelAgent(
            name=f"block_{run_id}",
            sub_agents=[Worker(name=n, run_id=run_id) for n in picked]
        )
        async for ev in parallel.run_async(ctx):
            yield ev

class Aggregator(BaseAgent):
    """Aggregates results from workers."""
    async def _run_async_impl(self, ctx):
        run_id = ctx.session.state.get("current_run")
        vals = [v for k, v in ctx.session.state.items()
                if run_id and k.startswith(f"result:{run_id}:")]
        yield Event(
            author=self.name,
            content=types.Content(role=self.name,
                   parts=[types.Part(text=f"Sum = {sum(vals)}")]),
            actions=EventActions(escalate=True)
        )

root_agent = SequentialAgent(
    name="root",
    sub_agents=[PlannerAndRunner(name="planner"), Aggregator(name="collector")]
)

✅ This simple pipeline dynamically fans out tasks to random workers, runs them in parallel, and aggregates the results — all in about 60 lines!

🏃 Running It

Launch your agent using the standard adk run command:

$ adk run .
Log setup complete: /tmp/agents_log/agent.20250427_122520.log
To access latest log: tail -F /tmp/agents_log/agent.latest.log
Running agent root, type exit to exit.

When prompted (user:), type anything (e.g., a, b) to trigger a new task round.

📒 Sample Output

Here's what you'll see:

user: a
[planner]: Run 84e9 tasks {'task:84e9:w0': 3, 'task:84e9:w1': 5}
[w1]: 5² = 25
[w0]: 3² = 9
[collector]: Sum = 34
user: b
[planner]: Run 35d1 tasks {'task:35d1:w1': 6, 'task:35d1:w0': 7, 'task:35d1:w2': 2}
[w1]: 6² = 36
[w0]: 7² = 49
[w2]: 2² = 4
[collector]: Sum = 89
user:

planner logs which workers received tasks
Each Worker logs its result
collector sums up the results per round

All steps are cleanly traceable thanks to ADK's event logs! 📈

🧩 Common Pitfalls and Best Practices

❌ Reusing Worker Instances

In ADK,

an agent instance can only have one parent (Multi-Agent Systems docs).

➡ Always create fresh Worker instances each time you build a new ParallelAgent.

🗄️ Designing session.state Safely

Since all agents share the same session.state,

prefix your keys (e.g., task:{run_id}:{worker_name}) to avoid collisions.

🏷️ Declaring ClassVar Properly

Because ADK agents are Pydantic models internally,

you must use ClassVar annotations for constants like POOL.

from typing import ClassVar
class PlannerAndRunner(BaseAgent):
    POOL: ClassVar[list[str]] = ["w0", "w1", "w2"]

Otherwise, Pydantic will treat them as model fields and raise an error.

📊 Comparing to Other Approaches

Approach	Limitations
Static ParallelAgent	Fixed set of children, no runtime flexibility
LLM `transfer_to_agent()`	Routing is flexible but serialized and LLM errors can break it
Manual asyncio.gather()	Breaks ADK's observability and session state management

✅ By dynamically building ParallelAgents,

you get the best of both worlds:

true concurrency + native ADK observability!

🌟 Extensions and Next Steps

Goal	Idea
Support multiple task types	Pass operation type to Workers
Summarize results with LLM	Replace Aggregator with an LlmAgent
LLM-assisted task dispatching	Use LlmAgent to select workers dynamically
Production deployment	Launch root agents with external triggers on Vertex AI

📝 Final Thoughts

Dynamic parallelism with Custom Agents

lets you build scalable, flexible, and highly performant workflows in ADK.

As long as you:

Handle Single-Parent Rule carefully
Design a clean session.state schema

you can achieve powerful architectures —

without losing any of ADK's built-in tracing and UI support.

If you're building dynamic, data-driven agent pipelines,

this technique belongs in your toolbox. 🛠️

🔽 Bonus Links

OmniSSHAgent: Simplifying SSH-Agent Communication on Windows

YAMASAKI Masahide — Tue, 01 Mar 2022 12:48:24 +0000

Managing SSH keys on Windows can be surprisingly complicated.

Unlike Linux, where ssh-agent communication is fairly standardized, Windows supports multiple different communication methods — leading to a tangled mess of tools and protocols.

In this post, we’ll walk through:

How SSH agents communicate with clients on Windows
Existing proxy tools for bridging communication gaps
The chaos this creates
How OmniSSHAgent cleans it all up

Let's dive in!

Understanding SSH-Agent Communication on Windows

On Windows, there are several different methods for an SSH client to talk to an SSH agent:

PuTTY (Pageant): Communicates using Windows Event Messages.
Cygwin/MSYS2 Unix Domain Socket: Simulates Unix-style sockets.
Windows Named Pipe: A native Windows IPC mechanism.
WSL1 Unix Domain Socket: For Windows Subsystem for Linux (WSL1).
WSL2 Unix Domain Socket: Native Unix sockets inside the WSL2 environment.

Because these methods are incompatible with each other, many proxy tools have been developed to bridge the gaps.

Existing Proxy Tools

Here are some popular tools that handle specific proxy needs:

wsl-ssh-agent

Proxies communication from a Windows Named Pipe to WSL1.

wsl-ssh-pageant

Bridges PuTTY (Pageant) and both Named Pipes and WSL1.

ssh-pageant

Bridges PuTTY (Pageant) to Cygwin/MSYS2 Unix domain sockets.

npiperelay + socat

Relays Named Pipe communication to WSL2 Unix sockets.

As you can see, each proxy only solves part of the problem.

You often need to juggle multiple tools depending on your setup.

Adding Keys to SSH-Agent

To use an SSH key, you typically need to load it into your agent.

ssh-add

The standard tool to add private keys to ssh-agent.
Available in OpenSSH environments.

KeePassXC

A popular password manager that also supports SSH key management.
Integrates with Pageant and Named Pipe-based agents.

The Chaos: A Visual Map

The reality of SSH-agent communication on Windows is messy.

Here’s a map illustrating just how chaotic it can get:

Multiple agents, multiple communication paths, and a lot of complexity to manage.

Enter OmniSSHAgent

OmniSSHAgent aims to simplify this mess.

It acts as a unified bridge, connecting all these communication methods under a single, consistent interface.

Here's what the world looks like when using OmniSSHAgent:

One tool. Clean, simple communication across all environments.

How to Get Started

Ready to untangle your SSH-agent setup?

👉 Check out the OmniSSHAgent GitHub page for installation and usage instructions.

Final Thoughts (My opinion)

In my opinion, OmniSSHAgent really shines if you're juggling multiple development environments (like WSL2 + regular Windows SSH + Pageant).

It cuts through a lot of the confusion, especially when working across different systems, and it’s lightweight enough that it won’t get in your way.

If you’re serious about dev work on Windows — it’s absolutely worth a look.