tessl/pypi-google-adk

Agent Development Kit - A flexible and modular framework for developing and deploying AI agents

—

Pending

Overview

Eval results

Files

Execution Engine

Name: tessl/pypi-google-adk
Author: tessl

Runners provide the execution environment for agents, supporting synchronous, asynchronous, and streaming execution modes with resource management and cleanup.

Capabilities

Core Runner Classes

Main execution engines that provide different execution modes and environments for agents.

class Runner:
    """Main execution engine with methods for different execution modes."""
    
    def __init__(
        self,
        *,
        app_name: str,
        agent: BaseAgent,
        plugins: Optional[List[BasePlugin]] = None,
        artifact_service: Optional[BaseArtifactService] = None,
        session_service: BaseSessionService,
        memory_service: Optional[BaseMemoryService] = None,
        credential_service: Optional[BaseCredentialService] = None,
    ):
        """
        Initialize the runner.
        
        Args:
            app_name (str): The application name of the runner
            agent (BaseAgent): The root agent to run
            plugins (Optional[List[BasePlugin]]): A list of plugins for the runner
            artifact_service (Optional[BaseArtifactService]): The artifact service for the runner
            session_service (BaseSessionService): The session service for the runner
            memory_service (Optional[BaseMemoryService]): The memory service for the runner
            credential_service (Optional[BaseCredentialService]): The credential service for the runner
        """
        pass
    
    def run(
        self,
        *,
        user_id: str,
        session_id: str,
        new_message: types.Content,
        run_config: RunConfig = RunConfig(),
    ) -> Generator[Event, None, None]:
        """
        Execute agent synchronously.
        
        Args:
            user_id (str): The user ID of the session
            session_id (str): The session ID of the session
            new_message (types.Content): A new message to append to the session
            run_config (RunConfig): The run config for the agent
            
        Yields:
            Event: The events generated by the agent
        """
        pass
    
    async def run_async(
        self,
        *,
        user_id: str,
        session_id: str,
        new_message: types.Content,
        state_delta: Optional[dict[str, Any]] = None,
        run_config: RunConfig = RunConfig(),
    ) -> AsyncGenerator[Event, None]:
        """
        Execute agent asynchronously.
        
        Args:
            user_id (str): The user ID of the session
            session_id (str): The session ID of the session
            new_message (types.Content): A new message to append to the session
            state_delta (Optional[dict[str, Any]]): Optional state changes
            run_config (RunConfig): The run config for the agent
            
        Yields:
            Event: The events generated by the agent
        """
        pass
    
    async def run_live(
        self,
        *,
        user_id: Optional[str] = None,
        session_id: Optional[str] = None,
        live_request_queue: LiveRequestQueue,
        run_config: RunConfig = RunConfig(),
        session: Optional[Session] = None,
    ) -> AsyncGenerator[Event, None]:
        """
        Execute agent in live/streaming mode.
        
        Args:
            user_id (Optional[str]): The user ID for the session. Required if session is None
            session_id (Optional[str]): The session ID for the session. Required if session is None  
            live_request_queue (LiveRequestQueue): The queue for live requests
            run_config (RunConfig): The run config for the agent
            session (Optional[Session]): The session to use (deprecated)
            
        Yields:
            Event: Events from the agent's live execution
        """
        pass
    
    async def close(self):
        """Clean up resources and close the runner."""
        pass

class InMemoryRunner(Runner):
    """In-memory runner for testing and development."""
    
    def __init__(
        self,
        agent: BaseAgent,
        *,
        app_name: str = 'InMemoryRunner',
        plugins: Optional[list[BasePlugin]] = None,
    ):
        """
        Initialize in-memory runner.
        
        Args:
            agent (BaseAgent): The root agent to run
            app_name (str): The application name of the runner. Defaults to 'InMemoryRunner'
            plugins (Optional[list[BasePlugin]]): A list of plugins for the runner
        """
        pass

Usage Examples

Basic Synchronous Execution

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

# Create agent and in-memory runner (recommended for simple usage)
agent = Agent(name="assistant", model="gemini-2.0-flash")
runner = InMemoryRunner(agent)

# Create message content
user_message = types.Content(parts=[types.Part(text="Hello, how can you help me?")])

# Execute synchronously and process events
for event in runner.run(
    user_id="user123",
    session_id="session456",
    new_message=user_message
):
    if event.content and not event.partial:
        print(event.content)

# Clean up
await runner.close()

Asynchronous Execution

import asyncio
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

async def main():
    agent = Agent(name="async_assistant", model="gemini-2.0-flash")
    runner = InMemoryRunner(agent)
    
    # Create message content
    user_message = types.Content(parts=[types.Part(text="What's the weather like?")])
    
    # Execute asynchronously and process events
    async for event in runner.run_async(
        user_id="user123", 
        session_id="session456",
        new_message=user_message
    ):
        if event.content and not event.partial:
            print(event.content)
    
    await runner.close()

# Run the async function
asyncio.run(main())

Live/Streaming Execution

from google.adk import Runner
from google.adk.agents import Agent

agent = Agent(name="streaming_assistant", model="gemini-2.0-flash")
runner = Runner()

# Execute in live mode for streaming responses
for chunk in runner.run_live(agent, input_text="Tell me a story"):
    print(chunk, end="", flush=True)

runner.close()

In-Memory Runner for Testing

from google.adk.runners import InMemoryRunner
from google.adk.agents import Agent

# Use in-memory runner for development/testing
runner = InMemoryRunner()
agent = Agent(name="test_agent", model="gemini-2.0-flash")

response = runner.run(agent, "Test input")
print(response)

runner.close()

Context Manager Pattern

from google.adk import Runner
from google.adk.agents import Agent

agent = Agent(name="context_agent", model="gemini-2.0-flash")

# Use runner as context manager for automatic cleanup
with Runner() as runner:
    response = runner.run(agent, "What can you do?")
    print(response)
# Runner is automatically closed

Batch Processing

from google.adk import Runner
from google.adk.agents import Agent

agent = Agent(name="batch_processor", model="gemini-2.0-flash")
runner = Runner()

inputs = [
    "Summarize this text...",
    "Translate this to French...",
    "Analyze this data..."
]

responses = []
for input_text in inputs:
    response = runner.run(agent, input_text)
    responses.append(response)

runner.close()

Error Handling

from google.adk import Runner
from google.adk.agents import Agent

agent = Agent(name="error_handling_agent", model="gemini-2.0-flash")
runner = Runner()

try:
    response = runner.run(agent, "Process this request", timeout=30)
    print(response)
except TimeoutError:
    print("Agent execution timed out")
except Exception as e:
    print(f"Error during execution: {e}")
finally:
    runner.close()

Install with Tessl CLI