PydanticAI

pip install netra-sdk pydantic-ai

from netra import Netra
from pydantic_ai import Agent
import os

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

# Create agent - automatically traced
agent = Agent(
    'openai:gpt-4',
    system_prompt='You are a helpful assistant.'
)

result = agent.run_sync('What is Pydantic?')
print(result.data)

from netra.decorators import agent, task
from netra import SpanWrapper
from pydantic_ai import Agent
from pydantic import BaseModel

class Response(BaseModel):
    answer: str
    confidence: float

@agent()
def create_qa_agent():
    span = SpanWrapper("create-agent").start()
    
    qa_agent = Agent(
        'openai:gpt-4',
        result_type=Response,
        system_prompt='Answer questions with confidence scores.'
    )
    
    span.end()
    return qa_agent

@task()
def query_agent(agent: Agent, question: str):
    query_span = SpanWrapper("query-agent", {
        "question": question
    }).start()
    
    result = agent.run_sync(question)
    query_span.set_attribute("answer", result.data.answer)
    query_span.set_attribute("confidence", result.data.confidence)
    query_span.end()
    
    return result.data

from netra.decorators import workflow
from netra import SpanWrapper
from pydantic_ai import Agent, RunContext

@workflow()
def agent_with_tools():
    agent = Agent('openai:gpt-4')
    
    @agent.tool
    def calculate(ctx: RunContext, x: float, y: float, op: str) -> float:
        """Perform mathematical operations."""
        tool_span = SpanWrapper("tool-calculate", {
            "operation": op,
            "x": x,
            "y": y
        }).start()
        
        if op == "add":
            result = x + y
        elif op == "multiply":
            result = x * y
        else:
            result = 0
        
        tool_span.set_attribute("result", result)
        tool_span.end()
        return result
    
    result = agent.run_sync('What is 15 + 27?')
    return result.data

from netra.decorators import task
from netra import SpanWrapper
from pydantic import BaseModel
from pydantic_ai import Agent

class Article(BaseModel):
    title: str
    summary: str
    tags: list[str]

@task()
def generate_article(topic: str) -> Article:
    span = SpanWrapper("generate-article", {
        "topic": topic
    }).start()
    
    agent = Agent(
        'openai:gpt-4',
        result_type=Article,
        system_prompt='Generate structured articles.'
    )
    
    result = agent.run_sync(f'Write about {topic}')
    
    span.set_attribute("article.title", result.data.title)
    span.set_attribute("article.tags", ",".join(result.data.tags))
    span.end()
    
    return result.data

from netra.decorators import task
from netra import SpanWrapper

@task()
async def stream_response(agent: Agent, prompt: str):
    stream_span = SpanWrapper("stream-response").start()
    
    async with agent.run_stream(prompt) as response:
        async for chunk in response.stream():
            print(chunk, end='', flush=True)
    
    stream_span.set_attribute("response", response.data)
    stream_span.end()
    
    return response.data

from netra.decorators import workflow
from netra import SpanWrapper
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class DatabaseDeps:
    connection_string: str

@workflow()
def agent_with_deps():
    agent = Agent(
        'openai:gpt-4',
        deps_type=DatabaseDeps
    )
    
    @agent.tool
    def query_db(ctx: RunContext[DatabaseDeps], query: str) -> str:
        """Query the database."""
        db_span = SpanWrapper("db-query", {
            "query": query
        }).start()
        
        # Use ctx.deps.connection_string
        result = f"Query result for: {query}"
        
        db_span.end()
        return result
    
    deps = DatabaseDeps(connection_string="postgresql://...")
    result = agent.run_sync('Get user data', deps=deps)
    return result.data