adk-fluent · Execution Modes Reference

1 The Six Execution Methods

.ask(prompt)

Sync

One-shot blocking call. Simplest way to get a response.

returns str

.ask_async(prompt)

Async

One-shot awaitable call. Safe inside any async context.

returns str

.stream(prompt)

Streaming

Async iterator yielding text chunks as they arrive.

yields str chunks

.events(prompt)

Streaming (raw)

Async iterator yielding raw ADK Event objects.

yields ADK Events

.map(prompts)

Sync Batch

Run multiple prompts in parallel. Blocks until all complete.

returns list[str]

.map_async(prompts)

Async Batch

Run multiple prompts concurrently. Awaitable.

returns list[str]

2 Environment Compatibility Matrix

Method	Script	Jupyter	FastAPI	adk web
.ask()	✓	✗RuntimeError	✗RuntimeError	✗
.ask_async()	✓*asyncio.run()	✓	✓	✓
.stream()	✗	✓	✓	✓
.events()	✗	✓	✓	✓
.map()	✓	✗RuntimeError	✗RuntimeError	✗
.map_async()	✓*asyncio.run()	✓	✓	✓

3 Execution Flow Diagrams

Sync .ask() Flow

Async .ask_async() Flow

Streaming .stream() / .events() Flow

4 The Golden Rule

Warning: Event Loop Conflict

SYNC methods (.ask(), .map()) call asyncio.run() internally. If you are already inside an async event loop (Jupyter, FastAPI, adk web), they will raise RuntimeError: This event loop is already running. Use the ASYNC variants (.ask_async(), .map_async()) instead.

5 Code Examples

Script (sync OK)

from adk_fluent import Agent

agent = Agent("helper", "gemini-2.5-flash")

result = agent.ask("Hello")  # Works
print(result)

Jupyter (must use async)

from adk_fluent import Agent

agent = Agent("helper", "gemini-2.5-flash")

# agent.ask("Hello")  # RuntimeError!

result = await agent.ask_async("Hello")  # Works

FastAPI (must use async)

from fastapi import FastAPI
from adk_fluent import Agent

app = FastAPI()
agent = Agent("helper", "gemini-2.5-flash")

@app.post("/chat")
async def chat(msg: str):
    return await agent.ask_async(msg)  # Works