CrewAI

Verified 2026-08-19 — PASS, 24s. Session persistence: wrapper-owned transcript.

CrewAI has no built-in cross-process session, so the wrapper keeps the transcript and replays it into the task description each turn.

---

The runner

import sys, os, json, pathlib
sys.path.insert(0, os.path.expanduser("~/agenttests"))
import azcfg, hetool

from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import tool

SESS_DIR = pathlib.Path(os.path.expanduser("~/agenttests/crewai/sessions"))
SESS_DIR.mkdir(parents=True, exist_ok=True)

@tool("get_working_folder")
def get_working_folder() -> str:
    """Return the agent's current working folder (absolute path)."""
    return hetool.get_working_folder()

@tool("run_shell")
def run_shell(command: str) -> str:
    """Run a shell command in the working folder and return its output."""
    return hetool.run_shell(command)

def load(sid):
    f = SESS_DIR / f"{sid}.json"
    return json.loads(f.read_text()) if f.exists() else []

def save(sid, transcript):
    (SESS_DIR / f"{sid}.json").write_text(json.dumps(transcript, indent=2))

def invoke(sid, message):
    transcript = load(sid)

    c = azcfg.load()
    llm = LLM(model="openai/" + c["model"], base_url=c["base_url"], api_key=c["api_key"])

    agent = Agent(
        role="Conversational assistant",
        goal="Answer the user, using the prior conversation for context and tools for operational questions.",
        backstory="You keep track of everything said earlier in this session.",
        llm=llm,
        tools=[get_working_folder, run_shell],
        verbose=False,
    )

    history = "\n".join(f'{t["role"]}: {t["content"]}' for t in transcript)
    desc = (
        "Here is the conversation so far in this session (may be empty):\n"
        f"{history or '(no prior messages)'}\n\n"
        f"New user message: {message}\n\n"
        "Reply concisely to the new user message."
    )
    task = Task(description=desc, expected_output="A concise reply.", agent=agent)
    crew = Crew(agents=[agent], tasks=[task], process=Process.sequential, verbose=False)

    reply = str(crew.kickoff()).strip()

    transcript.append({"role": "user", "content": message})
    transcript.append({"role": "assistant", "content": reply})
    save(sid, transcript)
    return reply

---

Engine definition

{
  "type": "process",
  "file": "/home/you/crewai/.venv/bin/python3",
  "nativeSession": true,
  "argsNew":    ["/home/you/he_adapter.py", "/home/you/crewai_session.py", "",            "{message}"],
  "argsResume": ["/home/you/he_adapter.py", "/home/you/crewai_session.py", "{sessionId}", "{message}"],
  "replyField": "result",
  "sessionIdField": "session_id",
  "approve": false,
  "workdir": "/home/you/work",
  "proxy": { "shape": "openai", "baseUrlEnv": "OPENAI_BASE_URL", "keyEnv": "OPENAI_API_KEY",
             "modelEnv": "OPENAI_MODEL", "model": "<route>|<provider model>" }
}

---

Adoption

1. Keep your existing crew, agents and tasks as they are.
2. Wrap the kickoff in invoke(sid, message).
3. Add transcript load/save keyed on sid, and replay it into the task description.
4. Read the route from azcfg.load() and build LLM(...) from it.
5. Seal the engine definition, point external.engine at it, restart.

---

Gotchas

model needs the openai/ prefix. CrewAI’s LLM routes on the provider prefix, so it is model="openai/" + c["model"], not the bare model id.

CrewAI prints its own tracing banner to stdout. The adapter redirects runner stdout to stderr — if you write your own adapter, do the same or the JSON envelope is unparseable.

Transcript replay is text, not structured turns. Tool calls from earlier turns are not replayed, only the final text. For long conversations, trim or summarise the history you replay.

verbose=False on both Agent and Crew keeps the output quiet; useful even with the stdout redirect in place.