CrewAI

pip install raindrop-crewai crewai

from raindrop_crewai import RaindropCrewAI
from crewai import Agent, Crew, Task

raindrop = RaindropCrewAI(
    api_key="your-write-key",
    user_id="user-123",
)
raindrop.setup()  # auto-patch all Crew.kickoff* methods

agent = Agent(
    role="Senior Researcher",
    goal="Find the most interesting facts about {topic}",
    backstory="You are an experienced researcher.",
)
task = Task(
    description="Identify 3 interesting facts about {topic}.",
    expected_output="A bulleted list of facts.",
    agent=agent,
)
crew = Crew(agents=[agent], tasks=[task])

result = crew.kickoff(inputs={"topic": "AI safety"})
print(result.raw)

raindrop.shutdown()

raindrop = RaindropCrewAI(
    api_key="your-write-key",            # Optional: omit to disable telemetry shipping
    user_id="user-123",          # Optional: default user ID for all events
    convo_id="convo-456",        # Optional: conversation/thread ID
    project_id="support-prod",   # Optional: route events to a specific project (slug)
    tracing_enabled=True,        # Optional: enable OTel CrewAI instrumentor (default True)
    bypass_otel_for_tools=True,  # Optional: forwarded to raindrop.init() (default True; no observable effect on CrewAI events)
    debug=False,                 # Optional: enable debug logging
)

# Auto-patch every Crew.kickoff* on the Crew class:
raindrop.setup()

# Or wrap a specific Crew instance (no global monkey-patch):
wrapped = raindrop.wrap(crew)

from raindrop_crewai import setup_crewai

raindrop = setup_crewai(api_key="your-write-key", user_id="user-123")
# Every Crew.kickoff* call is now automatically traced.

from raindrop_crewai import create_raindrop_crewai

raindrop = create_raindrop_crewai(api_key="your-write-key", user_id="user-123")
wrapped = raindrop.wrap(crew)
result = wrapped.kickoff(inputs={"topic": "AI safety"})
raindrop.shutdown()

raindrop = RaindropCrewAI(api_key="your-write-key", debug=True)

raindrop = RaindropCrewAI(
    api_key="your-write-key",
    project_id="support-prod",
)

from raindrop_crewai import RaindropCrewAI

# One long-lived wrapper per project, created at startup and reused.
rd_support = RaindropCrewAI(api_key="your-write-key", project_id="support-prod")
rd_billing = RaindropCrewAI(api_key="your-write-key", project_id="billing-prod")

# Wrap each crew with the corresponding wrapper.
support_crew = rd_support.wrap(support_crew)   # -> support-prod
billing_crew = rd_billing.wrap(billing_crew)   # -> billing-prod

support_crew.kickoff(inputs={"topic": "password reset"})
billing_crew.kickoff(inputs={"topic": "double charge"})

rd_support.flush()
rd_billing.flush()

from raindrop import Raindrop

client = Raindrop(api_key="your-write-key", project_id="support-prod")
rd = RaindropCrewAI(client=client)

from crewai import Agent, Crew, Task

researcher = Agent(role="Senior Researcher", goal="Research the topic", backstory="...")
analyst = Agent(role="Data Analyst", goal="Analyze findings", backstory="...")
writer = Agent(role="Technical Writer", goal="Write a clear report", backstory="...")

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process="sequential",  # or "hierarchical"
)

result = crew.kickoff(inputs={"topic": "quantum computing"})
# Captured properties include:
#   crewai.crew_name, crewai.process, crewai.agent_roles,
#   crewai.agent_count, crewai.task_count, crewai.tasks_output

from crewai.tools import tool

@tool("get_weather")
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 72F in {city}"

reporter = Agent(
    role="Weather Reporter",
    goal="Report the weather for the requested city",
    backstory="You always use the get_weather tool.",
    tools=[get_weather],
)

inputs = [
    {"topic": "AI safety"},
    {"topic": "quantum computing"},
    {"topic": "climate change"},
]

results = crew.kickoff_for_each(inputs=inputs)
# Captured as one ai_generation event with:
#   crewai.kickoff_type = "batch"
#   crewai.batch_size = 3
#   ai.usage.* aggregated across all runs

import asyncio

async def main():
    result = await crew.kickoff_async(inputs={"topic": "AI safety"})
    print(result.raw)

    results = await crew.kickoff_for_each_async(
        inputs=[{"topic": "AI"}, {"topic": "ML"}],
    )

asyncio.run(main())

raindrop = RaindropCrewAI(api_key="your-write-key", tracing_enabled=False)

raindrop.identify("user-123", {"name": "Alice", "plan": "pro"})

raindrop.track_signal(
    event_id="<event-id-from-dashboard-or-debug-log>",
    name="thumbs_up",
    signal_type="feedback",
    sentiment="POSITIVE",
    comment="Great answer!",
)

raindrop.flush()     # flush pending data
raindrop.shutdown()  # flush + release resources