CrewAI Integration

pip install mubit-crewai[crewai]

from mubit_crewai import MubitCrewMemory
from crewai import Crew, Process
 
memory = MubitCrewMemory(api_key="mbt_...", session_id="crew-run-1")
 
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    memory=memory.as_crew_memory(),
)
result = crew.kickoff(inputs={"topic": "AI safety"})
 
# Extended MuBit features
memory.checkpoint("Research phase complete")
memory.record_outcome("task-1", "success")
memory.handoff("researcher", "writer", "Here are findings", requested_action="execute")

from crewai import Agent, Task, Crew, Process
from mubit_crewai import MubitCrewMemory
 
memory = MubitCrewMemory(
    endpoint="https://api.mubit.ai",
    api_key="mbt_...",
    session_id="triage-001",
    agent_id="crewai-triage",
)
 
# Register agents for MAS coordination
for agent_def in [
    {"agent_id": "classifier", "role": "ticket-classifier"},
    {"agent_id": "researcher", "role": "solution-researcher"},
    {"agent_id": "responder",  "role": "response-drafter"},
]:
    memory.register_agent(**agent_def)
 
classifier = Agent(
    role="Support Ticket Classifier",
    goal="Classify tickets by severity and category",
    backstory="Experienced support lead who identifies severity and escalation triggers.",
    llm="openai/gpt-4o-mini",
)
 
researcher = Agent(
    role="Solution Researcher",
    goal="Find relevant past solutions from MuBit memory",
    backstory="Knowledge specialist who searches for similar tickets and resolution patterns.",
    llm="openai/gpt-4o-mini",
)
 
responder = Agent(
    role="Customer Response Drafter",
    goal="Draft empathetic, actionable replies",
    backstory="Senior success agent who turns frustrated customers into advocates.",
    llm="openai/gpt-4o-mini",
)
 
classify_task = Task(
    description="Classify this ticket: {ticket}",
    expected_output="Severity, category, key issues, escalation flags.",
    agent=classifier,
)
research_task = Task(
    description="Research solutions for: {ticket}",
    expected_output="Past cases, known solutions, systemic patterns.",
    agent=researcher,
)
respond_task = Task(
    description="Draft a response for: {ticket}",
    expected_output="Professional, empathetic customer response.",
    agent=responder,
)
 
crew = Crew(
    agents=[classifier, researcher, responder],
    tasks=[classify_task, research_task, respond_task],
    process=Process.sequential,
    memory=memory.as_crew_memory(),
)
 
result = crew.kickoff(inputs={"ticket": "Duplicate charge on Pro subscription..."})
 
# Post-run: handoffs, checkpoint, outcome
memory.handoff("classifier", "researcher", "Classification complete.", requested_action="continue")
memory.handoff("researcher", "responder", "Research complete.", requested_action="execute")
memory.checkpoint(snapshot="Triage complete.", label="triage-complete")
memory.record_outcome(reference_id="triage-001", outcome="success", rationale="Ticket resolved.")

memory.register_agent("...", role="...", read_scopes=[...], write_scopes=[...])
memory.handoff("from-agent", "to-agent", "...", requested_action="review")
memory.checkpoint("...", label="...")
memory.record_outcome("...", "success", rationale="...")
memory.surface_strategies()              # cluster recurring success lessons
memory.diagnose("error text", error_type="...", limit=10)  # failure-path lessons for debugging