> ## Documentation Index
> Fetch the complete documentation index at: https://docs.veadk.xyz/llms.txt
> Use this file to discover all available pages before exploring further.

# Runtime

A runtime organizes model calls, tool execution, and results. VeADK uses Google ADK by default, with `codex` and `piagent` available for coding-oriented execution. Check model settings, tools, and callbacks before switching

Complete [model configuration](/productions/veadk/preview/en/components/agent/model) first. Volcengine and BytePlus each require their own model endpoint and API Key

## Default runtime

Omitting `runtime` is equivalent to `runtime="adk"`, suitable for standard model calls, structured output, and tool workflows

```python main.py lines theme={null}
import asyncio
from veadk import Agent, Runner


agent = Agent(name="assistant", runtime="adk")

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain what an agent runtime does.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python main.py` to print the final response. Use `Runner.run_async` for event and streaming handling

## Parallel synchronous tool execution

Synchronous functions in the ADK runtime can block other async tasks. Set `tool_thread_pool_config` to run synchronous tools in worker threads, allowing concurrent execution when the model issues multiple calls in one turn

```python parallel_tools.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from google.adk.agents.run_config import ToolThreadPoolConfig

def lookup_stock(product: str) -> dict:
    """Return the demo stock count for a product."""
    return {"product": product, "stock": {"notebook": 12, "pen": 40}.get(product, 0)}

agent = Agent(
    name="stock_assistant",
    tools=[lookup_stock],
    tool_thread_pool_config=ToolThreadPoolConfig(max_workers=4),
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Check stock for notebook and pen.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python parallel_tools.py` to query both products. The model decides whether to issue calls together; the thread pool does not split tasks itself. Tools sharing mutable data or connections must support concurrent access

| Parameter | Type | Default | Description |
| :- | :- | :- | :- |
| `ToolThreadPoolConfig.max_workers` | `int` | `4` | Maximum workers, at least 1 |

`RunConfig.tool_thread_pool_config` takes precedence over the agent setting. It affects ADK synchronous tools, not Codex or PiAgent execution. See [code sandbox](/productions/veadk/preview/en/components/tools/code-sandbox#parallel-call-isolation) for parallel `run_code` session isolation and `VEADK_RUN_CODE_ISOLATE_PARALLEL_CALLS`

## Switch the execution backend

| `runtime` | Purpose | Preparation |
| :- | :- | :- |
| `adk` | Standard agent execution; default | No additional runtime installation |
| `codex` | Coding tasks through the Codex SDK | Install `veadk-python[codex]` |
| `piagent` | Execution through PiAgent RPC | Supply a binary or allow installation on first use |

### Use the Codex runtime

Install the optional SDK and CLI binary dependencies:

```bash lines theme={null}
pip install "veadk-python[codex]"
```

Model settings remain `model_name`, `model_api_base`, and `model_api_key`. Defaults use an isolated workspace, `workspace_write`, disabled network, and denied escalation. This example keeps those settings explicit

```python codex_runtime.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from veadk.runtime.codex import CodexRuntimeConfig

agent = Agent(
    name="coding_assistant",
    runtime="codex",
    codex_runtime_config=CodexRuntimeConfig(
        sandbox="workspace_write",
        approval_mode="deny_all",
        network_access=False,
    ),
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Calculate the sum of integers from 1 to 100.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python codex_runtime.py`. Register function and MCP tools through `Agent.tools`. Local skills in ADK `SkillToolset` can be passed to the runtime; see the compatibility table for legacy `skills_mode` restrictions

### Codex security configuration

`Agent.codex_runtime_config` accepts `CodexRuntimeConfig` or a matching dictionary

| Parameter | Type | Default | Description |
| :- | :- | :- | :- |
| `approval_mode` | `"deny_all" \| "auto_review"` | `"deny_all"` | `deny_all` rejects escalation; `auto_review` automatically accepts all such requests without human review |
| `sandbox` | `"read_only" \| "workspace_write" \| "full_access"` | `"workspace_write"` | Read-only, workspace writes, or full host access |
| `network_access` | `bool` | `False` | Controls network only under `workspace_write` |
| `workspace_root` | `str \| None` | `None` | Workspace root; otherwise an isolated temporary directory is created |
| `reuse_workspace` | `bool` | `False` | Reuses an explicitly configured workspace across invocations |
| `reasoning_effort` | `"minimal" \| "low" \| "medium" \| "high" \| "xhigh"` | `"medium"` | Reasoning effort; actual support depends on the model |
| `personality` | `"none" \| "friendly" \| "pragmatic"` | `"pragmatic"` | Response style |
| `max_tool_iterations` | `int` | `32` | Bridged ADK tool iteration budget for the whole turn, from 1 to 256; not a universal limit on native operations |
| `tool_timeout_seconds` | `float \| None` | `120.0` | Bridged tool timeout in seconds; must be positive, or `None` to omit this timeout |

<Warning>
  `auto_review` automatically approves escalation and file changes; it is not model-based review. `full_access` broadens host access, and `reuse_workspace=True` can share files across invocations. Use them only when explicitly needed in trusted environments. `network_access=False` cannot restrict `full_access`, so that combination is rejected. `read_only` also ignores this network switch
</Warning>

These environment variables override constructor settings:

| Variable | Default | Description |
| :- | :- | :- |
| `VEADK_CODEX_SANDBOX` | Unset | Overrides `sandbox` |
| `VEADK_CODEX_APPROVAL_MODE` | Unset | Overrides `approval_mode` |
| `VEADK_CODEX_WORKSPACE_ROOT` | Unset | Overrides `workspace_root` |
| `VEADK_CODEX_NETWORK_ACCESS` | Unset | Overrides `network_access`; `1`, `true`, `yes`, and `on` enable it, case-insensitively |

### Codex observability

Lifecycle notifications, function calls, and MCP calls become ADK events for sessions, traces, and frontends. Log fields such as `invocation_id`, `call_id`, `tool`, `status`, and `duration_ms` correlate operations. Token usage is exposed through `codex_event_type=token_usage` events

Runtime logs are not complete traces of every model call. Events, sessions, and configured exporters may contain task and tool content; control access and retention when exporting them

### Configure transient-error retries

| Variable | Type | Default | Description |
| :- | :- | :- | :- |
| `CODEX_SHIM_NUM_RETRIES` | `int` | `2` | Maximum retries for rate limits, server errors, overload, and timeouts; `0` disables retries |
| `CODEX_SHIM_TIMEOUT` | `float` | `0` | Backend model request timeout in seconds; `0` uses the client default |

### Use PiAgent

<Warning>
  PiAgent executes in a local working directory. Built-in tools may modify files or run commands. An isolated configuration directory is not an operating-system sandbox. Use trusted projects, a restricted environment, and appropriate tool limits
</Warning>

```python piagent_runtime.py lines theme={null}
import asyncio
from veadk import Agent, Runner


agent = Agent(name="coding_assistant", runtime="piagent")

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain the steps for checking a CSV file without modifying files.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python piagent_runtime.py`. VeADK checks `PIAGENT_BINARY`, then its managed cache, and downloads a Pi Release if neither is available. Download checksums are verified only when `PIAGENT_BINARY_SHA256` is set. Preinstall the binary and specify its path for offline and production environments

| Variable | Default | Description |
| :- | :- | :- |
| `PIAGENT_BINARY` | Unset | Installed executable |
| `PIAGENT_INSTALL_DIR` | `~/.cache/veadk/piagent` | Download and cache directory |
| `PIAGENT_BINARY_URL` | Release URL | Custom archive URL |
| `PIAGENT_BINARY_SHA256` | Unset | Expected archive SHA-256 |
| `PIAGENT_BINARY_VERSION` | `latest` | Version to download |
| `PIAGENT_BINARY_REPO` | `earendil-works/pi` | Release repository |
| `PIAGENT_BINARY_PLATFORM` | Current OS and architecture | Linux/macOS amd64 and arm64, plus Windows amd64 |
| `PIAGENT_AGENT_DIR` | Temporary directory per invocation | Isolated configuration and sessions; cannot be the real `~/.pi/agent` or its descendants |
| `PIAGENT_ALLOW_PARENT_PI_CODING_AGENT_DIR` | `false` | Allows the parent's `PI_CODING_AGENT_DIR` when no explicit directory is set; isolation checks still apply |
| `PIAGENT_WORKDIR` | Current directory | Task working directory |
| `PIAGENT_TIMEOUT_SECONDS` | `600` | Invocation timeout in seconds |
| `PIAGENT_PROVIDER_ID` | `veadk` | Provider identifier registered in Pi |
| `PIAGENT_MODEL_API` | `openai-completions` | Pi model API type |
| `PIAGENT_MODEL_API_KEY_ENV` | `VEADK_PI_MODEL_API_KEY` | Variable name used to pass the model Key to the subprocess |
| `PIAGENT_DISABLE_TOOLS` | `false` | Disables tools; bridge registration may re-enable them, so this is not an isolation guarantee |
| `PIAGENT_DISABLE_BUILTIN_TOOLS` | `false` | Disables Pi built-in tools |
| `PIAGENT_TOOL_ALLOWLIST` | Empty | Comma-separated built-in tool allowlist |
| `PIAGENT_EXCLUDE_TOOLS` | Empty | Comma-separated excluded tools |
| `PIAGENT_DISABLE_EXTENSION_DISCOVERY` | `true` | Disables automatic extension discovery |
| `PIAGENT_ENABLE_EXTENSION_DISCOVERY` | Unset | Enables discovery unless the matching `DISABLE` variable is set |
| `PIAGENT_DISABLE_SKILL_DISCOVERY` | `true` | Disables automatic skill discovery, not explicitly supplied skills |
| `PIAGENT_ENABLE_SKILL_DISCOVERY` | Unset | Enables discovery unless the matching `DISABLE` variable is set |
| `PIAGENT_PROJECT_TRUST` | `deny` | Project trust policy: `deny`, `approve`, or Pi's `default` |

### Runtime compatibility

Check these settings when switching to `codex` or `piagent`:

| Setting | External-runtime behavior |
| :- | :- |
| Custom `model`, `output_schema`, `planner`, agent-level `code_executor` | Unsupported; configuration fails |
| `include_contents="none"`, `enable_supervisor=True` | Unsupported; configuration fails |
| `generate_content_config` | Only `system_instruction` is supported; other explicit fields fail |
| `model_name` list, `model_fallbacks` | Uses only the primary model; ignores fallbacks |
| `model_provider` | Does not select a LiteLLM provider; requires a compatible endpoint |
| `enable_responses`, `enable_responses_cache` | Do not enable ADK's Ark Responses features |
| `model_extra_config` | Forwarded by Codex; unused by PiAgent |
| `knowledgebase`, `example_store` | Do not supply content automatically; use ADK or retrieve content and add it to instructions |
| Legacy `skills_mode`, `enable_skills_checklist` | Do not provide the full legacy toolset behavior; use ADK or an adapted native skill entry |
| `after_model_callback`, model traces | Operate on turn results; do not assume per-model-call callbacks and traces |
| `RunConfig.max_llm_calls` | Codex checks before calls; PiAgent counts completed calls and may stop one call beyond the limit |

## Agent transfer

`transfer_to_agent` lets the model delegate to another agent in the tree, which continues execution and returns results. ADK, Codex, and PiAgent support transfers

| Target | Condition |
| :- | :- |
| Child | Registered in `sub_agents`, excluding `single_turn` and `task` modes |
| Parent | A parent exists and `disallow_transfer_to_parent` is not set |
| Peer | Peer transfer is allowed and the target accepts transfers |

```python transfer.py lines theme={null}
import asyncio
from veadk import Agent, Runner


writer = Agent(
    name="writer",
    description="Rewrite supplied text as a concise announcement.",
    instruction="Rewrite the user's text as an announcement without adding facts.",
)
agent = Agent(
    name="coordinator",
    instruction="Transfer announcement-writing tasks to writer.",
    sub_agents=[writer],
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Write an announcement: the office closes at 17:00 on Friday.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python transfer.py` to obtain an announcement. Transfers require `Runner`; the target executes in the same invocation context, including its `output_key` behavior. Use the sequential workflow below when execution order must be fixed

## Output persistence

`output_key` stores the final response in session state under ADK, Codex, and PiAgent. It does not independently persist data to disk; survival across processes depends on [session storage](/productions/veadk/preview/en/components/session/index)

```python pipeline.py lines theme={null}
import asyncio
from google.adk.agents import SequentialAgent
from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory

planner = Agent(
    name="planner",
    instruction="Create an outline for the user's requested article.",
    output_key="plan",
)
writer = Agent(
    name="writer",
    instruction="Write the article using this outline: {plan}",
    output_key="draft",
)
pipeline = SequentialAgent(name="pipeline", sub_agents=[planner, writer])

async def main():
    runner = Runner(
        agent=pipeline, short_term_memory=ShortTermMemory(),
        app_name="writing_pipeline", user_id="demo-user",
    )
    print(await runner.run(
        messages="Write a short introduction to session state.", session_id="writing-demo"
    ))
    session = await runner.session_service.get_session(
        app_name="writing_pipeline", user_id="demo-user", session_id="writing-demo"
    )
    print(session.state["plan"])
    print(session.state["draft"])

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python pipeline.py`. The planner writes `plan`, the writer reads it through `{plan}`, and the article is stored in `draft`. The script prints the response and both state values for verification

## Model callbacks

Callbacks may be synchronous or asynchronous. ADK runs them around model calls; external runtimes run before/after callbacks around the whole turn, with plugin callbacks before agent callbacks. Returning `None` continues normal handling

| Callback | Behavior |
| :- | :- |
| `before_model_callback(callback_context, llm_request)` | Can change input, system instructions, and available tools; returning `LlmResponse` supplies a result and skips execution |
| `after_model_callback(callback_context, llm_response)` | Returning `LlmResponse` replaces the reply; external runtimes buffer and merge final text when this callback is configured |
| `on_model_error_callback(callback_context, llm_request, error)` | Returning `LlmResponse` supplies an error response; returning `None` propagates the exception |

Mutating a request does not add support for every ADK setting. The compatibility table still applies, including restrictions on `output_schema`

```python model_callback.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from google.adk.models.llm_response import LlmResponse
from google.genai import types

def fallback_on_error(callback_context, llm_request, error):
    return LlmResponse(content=types.Content(
        role="model",
        parts=[types.Part(text="The model is temporarily unavailable. Please retry later.")],
    ))

agent = Agent(name="assistant", on_model_error_callback=fallback_on_error)

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain session state.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python model_callback.py`. A successful model call returns its normal answer; a model failure returns the configured message. The callback does not handle every initialization, tool, or business error

### Convert PDF attachments to images

`pdf_to_images_before_model_callback` converts inline PDF bytes into images for models that support images but cannot read raw PDFs. PDF rendering dependencies are included in the default VeADK installation

Save a document suitable for sending to the model service as `example.pdf` in the current directory, select a vision-capable model, and run this script:

```python pdf_callback.py lines theme={null}
import asyncio
from pathlib import Path
from google.genai import types
from veadk import Agent, Runner
from veadk.utils.pdf_to_images import pdf_to_images_before_model_callback

async def main():
    agent = Agent(
        name="document_reader",
        before_model_callback=pdf_to_images_before_model_callback,
    )
    runner = Runner(agent=agent, app_name="pdf_demo", user_id="demo-user")
    await runner.session_service.create_session(
        app_name="pdf_demo", user_id="demo-user", session_id="pdf-session"
    )
    message = types.Content(role="user", parts=[
        types.Part(text="Summarize the document."),
        types.Part(inline_data=types.Blob(
            mime_type="application/pdf", data=Path("example.pdf").read_bytes()
        )),
    ])
    async for event in runner.run_async(
        user_id="demo-user", session_id="pdf-session", new_message=message
    ):
        if event.is_final_response() and event.content:
            print("".join(part.text or "" for part in event.content.parts or []))

if __name__ == "__main__":
    asyncio.run(main())
```

The default callback renders at most the first 10 pages of each PDF at scale `2.0`. To change these values, import `make_pdf_to_images_callback` from the same module and set `max_pages` and `scale`. More pages or a larger scale increase image volume, memory use, and model input usage

## Choose where each tool runs

`RuntimeProvider` controls individual tool calls, typically keeping the ADK flow while sending selected work to an application service. `DispatchRuntimeProvider` invokes your dispatch function, while `LocalRuntimeProvider` uses the original tool. To customize the full policy, subclass `RuntimeProvider` and implement `execute(tool_call)`

### Usage example

This runnable example simulates a service adapter locally. Dispatched stock is `12`; the original tool returning `0` is not executed again. It does not connect to a real remote service

```python dispatch.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from veadk.runtime import DispatchRuntimeProvider, ToolCall

def lookup_stock(product: str) -> dict:
    """Look up stock for a product."""
    return {"product": product, "stock": 0, "source": "local"}

async def dispatch_task(tool_call: ToolCall):
    # Demonstration adapter: replace this body with your service client.
    return {
        "product": tool_call.arguments["product"],
        "stock": 12,
        "source": "dispatch-demo",
    }

agent = Agent(
    name="stock_assistant",
    instruction="Use lookup_stock to answer stock questions.",
    tools=[lookup_stock],
)
runtime_provider = DispatchRuntimeProvider(
    dispatch_task, dispatchable_tools={"lookup_stock"}
)

async def main():
    print(await Runner(agent=agent, plugins=[runtime_provider]).run(
        messages='How many notebooks are in stock?', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

For a real service, replace `dispatch_task` with a client configured with authentication and timeouts, correlating requests with `tool_call.id`. Dispatch functions may be sync or async and should return a tool-compatible result. The application defines the service contract

Alternatively, pass `runtime_provider.before_tool_callback` to `Agent.before_tool_callback`. Do not register both entry points, which would intercept twice. MCP tools retain their own connections regardless of dispatch selection

### DispatchRuntimeProvider parameters

| Parameter | Type | Default | Description |
| :- | :- | :- | :- |
| `dispatch_task` | `Callable[[ToolCall], Any]` | — | Required. The function called for each dispatched tool call; receives a `ToolCall` and returns the tool result. May be synchronous or asynchronous |
| `dispatchable_tools` | `Collection[str] \| None` | `("bash",)` | Names of non-MCP tools to dispatch. Set to `None` to dispatch every non-MCP tool |
| `local_runtime` | `RuntimeProvider \| None` | `None` | Local provider for non-dispatched tools. Defaults to `LocalRuntimeProvider` |
| `name` | `str` | `"veadk_dispatch_runtime_provider"` | Plugin name |

### ToolCall fields

| Field / Property | Type | Description |
| :- | :- | :- |
| `name` | `str` | Tool name |
| `arguments` | `dict[str, Any]` | Arguments passed to the tool |
| `tool` | `BaseTool` | The ADK tool object, usable for local execution |
| `context` | `ToolContext` | The ADK tool context |
| `id` | `str` | Property. ADK function-call identifier; empty string when unavailable |
| `session_id` | `str` | Property. Current VeADK session identifier; empty string when unavailable |
| `RuntimeProvider(name=...)` requires a plugin name. `LocalRuntimeProvider` only has an optional `name`, defaulting to `"veadk_local_runtime_provider"` | | |

## Request processing

`run_processor` wraps the event stream of `Runner.run` for authentication, timing, cleanup, or event transformation. Precedence is the current `Runner.run(run_processor=...)`, the `Runner` constructor, the root agent, then the default pass-through processor. Calling `run_async` directly does not apply this wrapper

For authentication, pass `veadk.integrations.ve_identity.AuthRequestProcessor` to `Agent(run_processor=...)`. See [inbound authentication](/productions/veadk/preview/en/components/security/inbound) for identity setup

### Custom processors

Subclass `BaseRunProcessor` and implement `process_run(runner, message, **kwargs)`. This processor forwards events, records elapsed time, and closes the event stream on success, failure, or cancellation

```python run_processor.py lines theme={null}
import asyncio
from veadk import Agent, Runner
import time
from contextlib import aclosing
from veadk.processors.base_run_processor import BaseRunProcessor

class TimingProcessor(BaseRunProcessor):
    def process_run(self, runner, message, **kwargs):
        def decorator(event_generator):
            async def wrapper():
                started = time.perf_counter()
                try:
                    async with aclosing(event_generator()) as events:
                        async for event in events:
                            yield event
                finally:
                    print(f"Run elapsed: {time.perf_counter() - started:.2f}s")
            return wrapper
        return decorator

agent = Agent(name="assistant", run_processor=TimingProcessor())

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain what a runtime does.', session_id="runtime-demo"
    ))

if __name__ == "__main__":
    asyncio.run(main())
```

Run `python run_processor.py` to see the response and elapsed time. Before adding retries to a processor, determine whether a task has already produced external side effects to avoid repeating them