CrewAI

"""CrewAI on AgentOS: a Crew (`answer`) and a Flow that waits for a person before a large refund (`refund`)."""

from decimal import Decimal

from crewai import LLM, Agent, Crew, Task
from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

import agentos


def gateway_llm(inv: agentos.Invocation, role: str = "main") -> LLM:
    """CrewAI's LLM pointed at the AgentOS gateway, with this invocation's run token.

    provider="openai" makes CrewAI use its OpenAI-compatible client and send the model name exactly as
    agentos.yaml's config.models gives it (`openai/gpt-4o-mini`, `anthropic/claude-sonnet-4-5`, …).
    Build it inside the handler: the run token belongs to this invocation only."""
    return LLM(model=inv.llm.model(role), provider="openai", base_url=inv.llm.base_url,
               api_key=inv.llm.api_key, default_headers=inv.llm.headers, temperature=0)


# --- a Crew: runs start to finish, no pause -------------------------------------------------------------------

def answer(inv: agentos.Invocation):
    llm = gateway_llm(inv)
    researcher = Agent(role="Support researcher", goal="Find what the customer needs to know",
                       backstory="You know the product and its policies well.", llm=llm, verbose=False)
    writer = Agent(role="Support writer", goal="Write a short, friendly answer",
                   backstory="You turn research notes into replies customers understand.", llm=llm, verbose=False)
    research = Task(description="Research this customer question: {question}",
                    expected_output="Bullet-point notes", agent=researcher)
    write = Task(description="Write the reply to the customer from the notes.",
                 expected_output="A reply of at most 120 words", agent=writer, context=[research])
    result = Crew(agents=[researcher, writer], tasks=[research, write], verbose=False).kickoff(
        inputs={"question": inv.input["question"]})
    return {"answer": result.raw}


# --- a Flow: stops at a step boundary to ask a person, continues on the resume ----------------------------------

class RefundState(BaseModel):
    order_id: str = ""
    amount: str = ""
    customer_message: str = ""
    limit: str = "0"
    summary: str = ""            # written by the model in the first invocation, kept in run state
    decision: str = ""           # "", "approve", "edit" or "reject": filled in from the resume
    status: str = ""
    reply: str = ""


class RefundFlow(Flow[RefundState]):
    def __init__(self, inv: agentos.Invocation, **kwargs):
        super().__init__(**kwargs)
        self.inv = inv

    @start()
    def summarise(self):
        if self.state.summary:   # the resume: the model already did this step in the first invocation
            return
        agent = Agent(role="Refund assistant", goal="Summarise refund requests for the finance team",
                      backstory="You are precise and brief.", llm=gateway_llm(self.inv), verbose=False)
        self.state.summary = agent.kickoff(
            f"Summarise in one sentence why the customer wants a refund: {self.state.customer_message}").raw

    @router(summarise)
    def route(self):
        if self.state.decision == "reject":
            return "rejected"
        if self.state.decision or Decimal(self.state.amount) <= Decimal(self.state.limit):
            return "approved"
        return "needs_review"

    @listen("needs_review")
    def wait_for_review(self):
        self.state.status = "needs_review"   # the handler asks a person after the flow returns

    @listen("approved")
    def issue_refund(self):
        # Your side effect goes here; key it by run_id so a retried invocation can't refund twice.
        self.state.status = "refunded"
        self.state.reply = f"We refunded {self.state.amount} for order {self.state.order_id}."

    @listen("rejected")
    def decline(self):
        self.state.status = "declined"
        self.state.reply = "We couldn't approve this refund. A colleague will contact you."


def refund(inv: agentos.Invocation):
    if inv.resume is None:
        state = RefundState(**inv.input, limit=str(inv.config["params"]["refund_limit"]))
    else:
        state = RefundState(**inv.state.run.get("refund_flow"))
        decided = inv.resume.decided_items[0] if inv.resume.decided_items else None
        if decided is None or decided.action == "reject":
            state.decision = "reject"
        else:
            state.decision = decided.action
            state.amount = decided.data["amount"]   # a reviewer's edit wins

    flow = RefundFlow(inv)
    flow.kickoff(inputs=state.model_dump())

    if flow.state.status == "needs_review":
        inv.state.run.put("refund_flow", flow.state.model_dump())   # what the resume continues from
        inv.ask(f"Refund {flow.state.amount} for order {flow.state.order_id}?",
                [{"id": flow.state.order_id, "title": f"Order {flow.state.order_id} · {flow.state.amount}",
                  "data": {"amount": flow.state.amount, "summary": flow.state.summary}}],
                actions=["approve", "edit", "reject"],
                details="Approve, **edit** the amount, or reject.")
    return {"status": flow.state.status, "amount": flow.state.amount, "reply": flow.state.reply}


if __name__ == "__main__":
    agentos.serve({"answer": answer, "refund": refund}, port=8080)

name: crewai-support
title: CrewAI support agent
description: A CrewAI crew that answers support questions, and a CrewAI Flow that asks a person before a large refund.
runtime: stateless
port: 8080
capabilities:
  answer:
    description: A crew of two agents researches a support question and writes the answer.
    input:
      type: object
      properties: {question: {type: string}}
      required: [question]
    output:
      type: object
      properties: {answer: {type: string}}
      required: [answer]
    timeout: 5m
  refund:
    description: Drafts a refund; refunds above the limit wait for a person's approval.
    input:
      type: object
      properties:
        order_id: {type: string}
        amount: {type: string}
        customer_message: {type: string}
      required: [order_id, amount, customer_message]
    output:
      type: object
      properties:
        status: {type: string}
        amount: {type: string}
        reply: {type: string}
      required: [status, reply]
    timeout: 5m
config:
  models: {main: openai/gpt-4o-mini}
  params: {refund_limit: "100.00"}

FROM python:3.12-slim
WORKDIR /app
# CrewAI writes its storage under HOME and phones home unless told not to
ENV HOME=/tmp CREWAI_STORAGE_DIR=/tmp/crewai CREWAI_DISABLE_TELEMETRY=true CREWAI_TRACING_ENABLED=false \
    OTEL_SDK_DISABLED=true PYTHONUNBUFFERED=1
COPY requirements.txt .
COPY vendor/ vendor/
RUN pip install --no-cache-dir -r requirements.txt vendor/*.whl
COPY app.py .
USER nobody
EXPOSE 8080
CMD ["python", "app.py"]

crewai==1.15.30