CrewAI

pip install netra-sdk crewai crewai-tools

from netra import Netra
from crewai import Agent, Task, Crew
import os

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

# Define agents - automatically traced
researcher = Agent(
    role="Researcher",
    goal="Research and analyze information",
    backstory="Expert researcher with attention to detail"
)

writer = Agent(
    role="Writer",
    goal="Write engaging content",
    backstory="Creative writer with storytelling skills"
)

# Create tasks
research_task = Task(
    description="Research AI trends",
    agent=researcher,
    expected_output="Research report"
)

write_task = Task(
    description="Write article based on research",
    agent=writer,
    expected_output="Article"
)

# Create crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task]
)

result = crew.kickoff()

from netra.decorators import workflow, agent, task
from netra import SpanWrapper

@workflow()
def research_workflow(topic: str):
    span = SpanWrapper("research-workflow", {
        "topic": topic
    }).start()
    
    # Create specialized agents
    researcher = Agent(
        role="Researcher",
        goal=f"Research {topic}",
        backstory="Expert researcher"
    )
    
    analyst = Agent(
        role="Analyst",
        goal="Analyze research data",
        backstory="Data analyst"
    )
    
    # Define tasks
    research = Task(
        description=f"Research {topic}",
        agent=researcher,
        expected_output="Research findings"
    )
    
    analysis = Task(
        description="Analyze research",
        agent=analyst,
        expected_output="Analysis report"
    )
    
    crew = Crew(
        agents=[researcher, analyst],
        tasks=[research, analysis],
        verbose=True
    )
    
    result = crew.kickoff()
    span.set_attribute("result.status", "completed")
    span.end()
    
    return result

from netra.decorators import task
from netra import SpanWrapper

@task()
def execute_agent_task(agent: Agent, task_description: str):
    task_span = SpanWrapper("agent-task", {
        "agent.role": agent.role,
        "task.description": task_description
    }).start()
    
    try:
        task = Task(
            description=task_description,
            agent=agent,
            expected_output="Task result"
        )
        
        result = agent.execute_task(task)
        task_span.set_attribute("task.result", str(result))
        task_span.end()
        
        return result
    except Exception as e:
        task_span.set_attribute("error", str(e))
        task_span.set_status(code=1, message=str(e))
        task_span.end()
        raise

from netra.decorators import agent
from netra import SpanWrapper
from crewai import Process

@agent()
def run_crew_with_config(agents: list[Agent], tasks: list[Task]):
    config_span = SpanWrapper("crew-config", {
        "agents.count": len(agents),
        "tasks.count": len(tasks)
    }).start()
    
    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
        memory=True
    )
    
    config_span.end()
    
    # Execute crew
    exec_span = SpanWrapper("crew-execution").start()
    result = crew.kickoff()
    exec_span.set_attribute("execution.status", "completed")
    exec_span.end()
    
    return result

from netra.decorators import workflow
from netra import SessionManager

@workflow()
def interactive_crew_session(user_id: str, queries: list[str]):
    session_manager = SessionManager()
    session_manager.start_session(user_id=user_id)
    
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task]
    )
    
    results = []
    for query in queries:
        conversation_id = session_manager.start_conversation()
        
        result = crew.kickoff(inputs={"query": query})
        results.append(result)
        
        session_manager.end_conversation()
    
    session_manager.end_session()
    return results

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

# Enable specific instruments
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True,
    instruments={InstrumentSet.OPENAI}  # CrewAI uses OpenAI
)