Runtime

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())

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())

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())

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())

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())

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())

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())

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())

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())

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())