Google ADK Integration

1 Install

pip install keylightdigital-nexus google-adk

Requires Python 3.9+.

2 Create an API Key

Go to /dashboard/keys and create a new API key.

export NEXUS_API_KEY="nxs_your_api_key_here"

3 Initialize the Client

import os
from nexus_client import NexusClient

nexus = NexusClient(
    api_key=os.environ["NEXUS_API_KEY"],
    agent_id="google-adk-agent",
)

4 Trace Agent Runs

Wrap your ADK Runner.run_async() call with a Nexus trace. One trace = one complete agent invocation.

import asyncio
import os
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from nexus_client import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"], agent_id="research-agent")

agent = Agent(
    name="research_agent",
    model="gemini-2.0-flash-exp",
    instruction="You are a helpful research assistant.",
)

async def run_agent(user_query: str) -> str:
    trace = nexus.start_trace(
        name="adk: " + user_query[:60],
        metadata={"model": "gemini-2.0-flash-exp"},
    )
    try:
        session_service = InMemorySessionService()
        runner = Runner(agent=agent, app_name="app", session_service=session_service)
        session = session_service.create_session(app_name="app", user_id="u1")
        from google.genai import types
        content = types.Content(role="user", parts=[types.Part(text=user_query)])
        final_response = ""
        async for event in runner.run_async(
            user_id="u1", session_id=session.id, new_message=content
        ):
            if event.is_final_response() and event.content:
                for part in event.content.parts:
                    if part.text:
                        final_response += part.text
        trace.end(status="success")
        return final_response
    except Exception as e:
        trace.end(status="error")
        raise

5 Trace Tool Calls

Intercept ADK events in the run loop to capture individual tool call spans:

async def run_with_tool_tracing(query: str) -> str:
    trace = nexus.start_trace(name="adk-tools: " + query[:60])
    try:
        session_service = InMemorySessionService()
        runner = Runner(agent=agent, app_name="app", session_service=session_service)
        session = session_service.create_session(app_name="app", user_id="u1")
        from google.genai import types
        content = types.Content(role="user", parts=[types.Part(text=query)])
        final_response = ""
        async for event in runner.run_async(
            user_id="u1", session_id=session.id, new_message=content
        ):
            # Capture tool call events from the ADK event stream
            if hasattr(event, "tool_call") and event.tool_call:
                span = trace.add_span(
                    name="tool:" + event.tool_call.name,
                    input=dict(event.tool_call.args),
                )
                span.end(status="ok")
            if event.is_final_response() and event.content:
                for part in event.content.parts:
                    if part.text:
                        final_response += part.text
        trace.end(status="success")
        return final_response
    except Exception as e:
        trace.end(status="error")
        raise

6 Multi-Agent Pipelines

For pipelines orchestrating multiple ADK agents, use a parent trace with child spans per sub-agent:

async def run_multi_agent_pipeline(task: str) -> str:
    trace = nexus.start_trace(
        name="pipeline: " + task[:60],
        metadata={"pipeline": "research-and-summarize"},
    )
    try:
        r_span = trace.add_span(name="sub-agent:research", input={"task": task})
        research_result = await run_research_agent(task)
        r_span.end(status="ok", output={"length": len(research_result)})

        s_span = trace.add_span(
            name="sub-agent:summarize",
            input={"content_length": len(research_result)},
        )
        summary = await run_summary_agent(research_result)
        s_span.end(status="ok", output={"length": len(summary)})

        trace.end(status="success")
        return summary
    except Exception as e:
        trace.end(status="error")
        raise