Your First ADK Agent Harness: Hello World with agents-cli

Building a production-ready AI agent requires more than sending prompts to an LLM; it requires a structured harness that manages execution flow, handles state persistence, and provides tool integration.

In this step-by-step tutorial, we will build a complete, runnable agent harness using the Google Agent Development Kit (ADK) in Python. We’ll construct a customer support agent that utilizes custom function tools, session callbacks, and ADK’s core Runner abstraction.

---

Project Setup

We’ll use uv for fast, reproducible Python project management. Open your terminal and initialize a new project:

uv init hello-adk
cd hello-adk
uv add google-adk google-genai

Environment Configuration

Ensure your Google GenAI API credentials are available in your environment:

export GEMINI_API_KEY="your-api-key-here"

Alternatively, if running within Google Cloud, configure Application Default Credentials (ADC) via gcloud auth application-default login.

---

Designing the Agent Harness

We will create our harness inside main.py. The file structure is divided into three functional parts:

1. A Custom Function Tool for fetching user data.
2. A Lifecycle Callback for managing turn counters in session state.
3. The Agent & Runner Construction to execute queries.

1. Define a Custom Function Tool

Google ADK utilizes Python docstrings and type hints to automatically extract an OpenAPI JSON schema for Gemini.

import asyncio
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.agents.callback_context import CallbackContext

def get_user_account_status(user_id: str) -> dict:
    """Retrieves account tier and loyalty balance for a given user ID.

    Args:
        user_id: The unique identifier for the customer (e.g. 'USR-101').
    """
    accounts = {
        "USR-101": {"tier": "Gold", "points": 14500, "status": "Active"},
        "USR-102": {"tier": "Standard", "points": 200, "status": "Active"},
    }
    return accounts.get(user_id, {"tier": "Unknown", "points": 0, "status": "NotFound"})

# Wrap the native python function into an ADK tool
account_tool = FunctionTool(func=get_user_account_status)

---

2. Implement a Session Callback

Callbacks let developers inspect or alter state during execution. Here, we register a before_agent_callback to seed and track invocation counts across conversation turns:

async def initialize_session_state(callback_context: CallbackContext) -> None:
    """Callback that runs before the agent invokes the model on each turn."""
    state = callback_context.state
    if "invocation_count" not in state:
        state["invocation_count"] = 0
    
    state["invocation_count"] += 1
    print(f"[HARNESS LOG] Executing turn #{state['invocation_count']}")

---

3. Assemble the Agent and Runner

Next, instantiate the Agent definition and couple it with an InMemorySessionService inside an ADK Runner:

support_agent = Agent(
    name="customer_support_agent",
    model="gemini-2.5-flash",
    instruction=(
        "You are an automated customer support assistant. "
        "Use the `get_user_account_status` tool to fetch customer information when asked."
    ),
    tools=[account_tool],
    before_agent_callback=initialize_session_state,
)

async def main():
    from google.adk.runners import Runner
    from google.adk.sessions import InMemorySessionService

    # 1. Initialize the session service
    session_service = InMemorySessionService()

    # 2. Attach agent and session service to the runner
    runner = Runner(agent=support_agent, session_service=session_service)

    # 3. Execute a query within a specific session context
    session_id = "sess-alpha-001"
    prompt = "Can you check the status for customer USR-101?"

    print(f"User Prompt: {prompt}\n")
    response = await runner.run_async(session_id=session_id, message=prompt)

    print("\n--- Agent Response ---")
    print(response.text)

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

---

Execution & Output

Run your newly constructed agent harness with uv:

uv run python main.py

Expected Output

User Prompt: Can you check the status for customer USR-101?

[HARNESS LOG] Executing turn #1

--- Agent Response ---
Customer USR-101 is currently on the Gold tier with 14,500 loyalty points, and their account status is Active.

---

Architectural Breakdown: What Happened Under the Hood?

When runner.run_async() was invoked, the ADK harness executed the following sequence:

sequenceDiagram
    participant User
    participant Runner
    participant Callback
    participant Gemini Model
    participant Function Tool

    User->>Runner: Submit message ("check status USR-101")
    Runner->>Callback: Trigger before_agent_callback
    Note over Callback: Increment invocation_count in session state
    Runner->>Gemini Model: Send system instructions + user message + tool schemas
    Gemini Model-->>Runner: Request function call: get_user_account_status(user_id="USR-101")
    Runner->>Function Tool: Execute get_user_account_status("USR-101")
    Function Tool-->>Runner: Return account data dict
    Runner->>Gemini Model: Submit function call response payload
    Gemini Model-->>Runner: Generate final natural language summary
    Runner-->>User: Return response text

1. Automatic Schema Synthesis: ADK parsed parameter types and the docstring of get_user_account_status into an OpenAPI tool schema.
2. Lifecycle Interception: initialize_session_state executed before the LLM request, managing session memory.
3. Loop Control: The Runner sent the prompt to Gemini 2.5 Flash, recognized the function call response, executed the Python function locally, and re-submitted the result back to Gemini for final synthesis.

---

Next Steps

While InMemorySessionService is ideal for local testing and unit tests, production deployments require persistent backends (such as SQLite or PostgreSQL) via DatabaseSessionService so that state survives server restarts.