Google ADK

pip install netra-sdk google-adk

from netra import Netra
from google.adk import Agent
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Define agent - automatically traced
agent = Agent(
    name="assistant",
    model="gemini-1.5-pro",
    instructions="You are a helpful assistant"
)

# Run agent
response = await agent.run("What is Google ADK?")
print(response.output)

from netra.decorators import task
from netra import SpanWrapper
from google.adk import Agent

@task()
async def run_agent(prompt: str) -> str:
    span = SpanWrapper("adk-agent", {
        "prompt": prompt
    }).start()
    
    agent = Agent(
        name="assistant",
        model="gemini-1.5-pro",
        instructions="You are a helpful assistant"
    )
    
    response = await agent.run(prompt)
    
    span.set_attribute("response", response.output)
    span.end()
    
    return response.output

from netra.decorators import agent
from netra import SpanWrapper
from google.adk import Agent, Tool

def get_weather(location: str) -> str:
    """Get weather information for a location."""
    return f"Weather in {location}: Sunny, 72°F"

@agent()
async def agent_with_tools(query: str):
    tool_span = SpanWrapper("adk-agent-tools", {
        "query": query
    }).start()
    
    # Define tool
    weather_tool = Tool(
        name="get_weather",
        description="Get weather information",
        function=get_weather
    )
    
    # Create agent with tool
    assistant = Agent(
        name="weather-agent",
        model="gemini-1.5-pro",
        instructions="Help users with weather information",
        tools=[weather_tool]
    )
    
    response = await assistant.run(query)
    tool_span.set_attribute("response", response.output)
    tool_span.end()
    
    return response.output

from netra.decorators import workflow
from netra import SpanWrapper
from google.adk import Agent

@workflow()
async def multi_agent_system(task: str):
    workflow_span = SpanWrapper("multi-agent-workflow", {
        "task": task
    }).start()
    
    # Research agent
    researcher = Agent(
        name="researcher",
        model="gemini-1.5-pro",
        instructions="Research and gather information"
    )
    
    # Writer agent
    writer = Agent(
        name="writer",
        model="gemini-1.5-pro",
        instructions="Write based on research"
    )
    
    # Execute agents in sequence
    research_result = await researcher.run(f"Research: {task}")
    final_result = await writer.run(
        f"Write based on: {research_result.output}"
    )
    
    workflow_span.set_attribute("result", final_result.output)
    workflow_span.end()
    
    return final_result.output

from netra.decorators import task
from netra import SpanWrapper

@task()
async def stream_agent_response(prompt: str):
    stream_span = SpanWrapper("adk-stream").start()
    
    agent = Agent(
        name="assistant",
        model="gemini-1.5-pro",
        instructions="You are a helpful assistant"
    )
    
    full_response = ""
    async for chunk in agent.stream(prompt):
        print(chunk, end='', flush=True)
        full_response += chunk
    
    stream_span.set_attribute("response", full_response)
    stream_span.end()
    
    return full_response

from netra.decorators import agent
from netra import SpanWrapper
from google.adk import Agent, Memory

@agent()
async def stateful_agent(user_id: str, message: str):
    state_span = SpanWrapper("adk-stateful", {
        "user_id": user_id,
        "message": message
    }).start()
    
    # Create agent with memory
    assistant = Agent(
        name="assistant",
        model="gemini-1.5-pro",
        instructions="Remember context from previous interactions",
        memory=Memory(user_id=user_id)
    )
    
    response = await assistant.run(message)
    state_span.set_attribute("response", response.output)
    state_span.end()
    
    return response.output

from netra import Netra
from netra.instrumentation.instruments import InstrumentSet

# Enable Google ADK instrumentation
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True,
    instruments={InstrumentSet.ADK}
)