Migrating the Agno workflow engine to Claude Agent SDK
A staged plan to move AgencyCore's AI orchestration off the Agno runtime onto Anthropic's first-party Agent SDK & Managed Agents - with an honest line drawn between what becomes a managed agent and what stays a deterministic pipeline.
Executive summary
tl;dr AgencyCore runs three multi-step AI workflows (Sonar, Headhunter, Envoy) plus ~26 single-shot agents and one conversational chat toolkit, all on the Agno framework against OpenAI gpt-5.4. This document specifies the migration to Anthropic's Agent SDK. The central decision: not everything should become a Managed Agent - we tier the surface by control-flow shape and route each construct to the cheapest primitive that fits.
Managed Agents are for open-ended, model-driven exploration with server-hosted tool execution. The chat agent fits that exactly. The three workflows are deterministic DAGs - they keep their Python orchestration and swap Agno agent calls for direct Anthropic SDK messages.parse() calls. Force-fitting a DAG into a model-driven agent loop would forfeit determinism, parallel fan-out, and cost control.
Current state · the Agno surface
inventory Everything inherits from BaseAgnoWorkflow or instantiates an Agno Agent. Models are OpenAI via OpenAIResponses. Web tooling wraps Agno's ExaTools / FirecrawlTools.
Multi-step pipelines
Celery-driven, SSE-streamed via Redis + Supabase workflow_run_logs. Deterministic step graphs with parallel fan-out.
Structured extractors
Module-level singletons with Pydantic output_schema, a PromptInjectionGuardrail, run via agent.arun(). No tools, one round-trip each.
Chat agent
Multi-turn, tool-calling agent built per-request via factory. Registers AcCliToolkit (3 functions) over the ac CLI with allowlist policy.
CompanySearchWorkflow · SonarHeadhunterWorkflowEnvoyEmailDraftWorkflowTarget · Claude Agent SDK primitives
vocabulary The Agent SDK exposes three surfaces over one endpoint. The mandatory flow for Managed Agents is Agent (once) → Session (every run). model/system/tools live on the persisted, versioned agent - never on the session.
Claude API
One request, one response - classification, extraction, summarization. With messages.parse() + output_config.format for validated Pydantic output.
API + tool-use loop
You own the orchestration loop in Python. Tools defined with @beta_tool or raw JSON schema; the runner handles the call->execute->feed-back cycle.
Managed Agents
Anthropic runs the loop & hosts a per-session container. SSE event stream, persisted versioned config, Skills + MCP, custom tools executed host-side.
agents.create() once. Versioned config: model, system, tools, mcp_servers, skills, multiagent roster.
Per-run. References agent by id, attaches an environment + resources. Produces an SSE event stream.
agent_toolset_20260401 (bash/read/write/edit/glob/grep/web_*), mcp_toolset, and custom (host-executed).
Workspace-scoped memory stores (FUSE-mounted). user.define_outcome rubric → grade → revise loop.
The tiering thesis
decisionThe Agent SDK's own decision tree routes deterministic multi-step pipelines and single-shot extraction away from Managed Agents. We follow it. Below, each construct class is matched to a target tier by its control-flow shape.
~26 Agents → messages.parse()
Single round-trip, structured output, no tools. A Managed Agent here would add a container, an event stream, and loop latency for zero benefit. Map 1:1 to a direct Claude API call with a Pydantic schema.
strategist · auditor · qualifier · extractor · writers · validators · demo gens · csv mapper
3 Workflows → API tool-use
Keep the Python/Celery DAG and its parallel fan-out, telemetry, and ordering guarantees. Replace each agno Agent.arun() node with an Anthropic SDK call. Optionally promote to a Managed Agent per-run session only where a step genuinely needs model-driven exploration.
CompanySearch · Headhunter · EnvoyEmailDraft
Chat agent → Managed Agents
The natural fit: conversational, multi-turn, model decides which ac command to run. Maps to a persisted agent + custom tools (host-side CLI exec) + per-session SSE streaming, replacing the bespoke chat runner entirely.
AcCliToolkit → custom tools · allowlist policy → tool permission policy
Wrapping the 7-step Sonar DAG in a single Managed Agent user.define_outcome loop. It would replace deterministic, observable, parallel steps with an opaque model-driven trajectory - slower, costlier, and harder to debug. Reserve Managed Agents for where the trajectory is genuinely unknown ahead of time.
Construct mapping
1:1 tableSelect a tier to see how each Agno construct translates.
| Agno construct | Agent SDK target | Notes | |
|---|---|---|---|
Agno Agent + output_schema | → | client.messages.parse(output_config=…) | Pydantic model passes straight through; strict-schema validation server-side. |
OpenAIResponses("gpt-5.4") | → | model="claude-opus-4-8" | Adaptive thinking + effort replace temperature tuning. |
PromptInjectionGuardrail | → | pre-call validator + system prompt | Port the validator; Claude refuses more appropriately by default. |
| module-level singleton agent | → | cached client + frozen system prompt | System prompt cached via cache_control for ~90% input savings. |
batch helper (run_in_batches) | → | messages.batches or asyncio gather | Use Batches API for non-latency-sensitive scoring (50% cost). |
Per-workload designs
concreteStrategist / Auditor / Qualifier
One call, validated output, frozen cached system prompt. The dominant pattern - ~26 instances collapse to this shape.
▶company_search_strategist.pypython
from anthropic import Anthropic
from pydantic import BaseModel
class TacticalPlan(BaseModel):
industries: list[str]
keywords: list[str]
locations: list[str]
# module-level singleton client; system prompt is cached
client = Anthropic()
async def plan(query: str) -> TacticalPlan:
r = await client.messages.parse(
model="claude-opus-4-8",
max_tokens=4096,
thinking={"type": "adaptive"},
output_config={"effort": "low"},
system=[{"type":"text","text":STRATEGIST_PROMPT,
"cache_control":{"type":"ephemeral"}}],
messages=[{"role":"user","content":query}],
output_format=TacticalPlan,
)
return r.parsed_outputHeadhunter researcher node
The Celery orchestrator is unchanged; the LLM node uses the SDK tool runner with Exa as a host-side tool. Fan-out / dedup / ordering stay in Python.
▶researcher_node.pypython
from anthropic import beta_tool
@beta_tool
def exa_search(q: str, n: int = 10) -> str:
"""Search the web via Exa for candidate profiles."""
return sanitized_exa.run(q, num_results=n) # host-side, keyed
async def research_company(company) -> list[Candidate]:
runner = client.beta.messages.tool_runner(
model="claude-opus-4-8", max_tokens=16000,
tools=[exa_search],
messages=[{"role":"user",
"content": research_prompt(company)}],
)
for msg in runner: ... # loop owned by SDK
return parse_candidates(msg)
# orchestrator still drives conc=6 fan-out across companiesAcCliToolkit → custom tools, host-executed
Created once at setup and stored by id. The session streams events; when the agent calls run_ac_command, our orchestrator runs the CLI host-side and replies with user.custom_tool_result - the container never holds Supabase credentials.
▶chat_agent_setup.py & runtimepython · managed agents (beta)
# ── ONE-TIME SETUP — store agent.id in config ──────────────
agent = client.beta.agents.create(
name="AgencyCore Chat",
model="claude-opus-4-8",
system=CHAT_SYSTEM_PROMPT,
tools=[
{"type":"custom","name":"run_ac_command",
"description":"Run one allowlisted ac CLI command.",
"input_schema":{"type":"object",
"properties":{"cmd":{"type":"string"}},"required":["cmd"]}},
# run_ac_commands, run_ac_commands_write …
],
)
# ── RUNTIME — every chat turn ──────────────────────────────
session = client.beta.sessions.create(
agent=AGENT_ID, environment_id=ENV_ID)
stream = client.beta.sessions.events.stream(session.id) # stream-first
client.beta.sessions.events.send(session.id, events=[
{"type":"user.message",
"content":[{"type":"text","text": user_msg}]}])
for ev in stream:
if ev.type == "agent.custom_tool_use":
out = run_ac_host_side(ev.input["cmd"]) # allowlist + keys here
client.beta.sessions.events.send(session.id, events=[{
"type":"user.custom_tool_result",
"custom_tool_use_id": ev.id,
"content":[{"type":"text","text": out}]}])
elif ev.type == "session.status_idle" \
and ev.stop_reason.type != "requires_action":
breakalways_ask → the session idles on requires_action and we answer with user.tool_confirmation, preserving today's write-confirmation gate.Cross-cutting concerns
platformgpt-5.4 → opus-4-8
temperature removed - steer by prompt + effort. Adaptive thinking on intelligence-sensitive nodes; low effort for cheap extractors.
Web tools host-side
Exa / Firecrawl / Hunter become @beta_tool fns (Tier 2) or custom tools (Tier 3). Keys never enter any sandbox.
Pydantic → format
Every output_schema maps to messages.parse() with output_config.format. Strict mode guarantees valid JSON.
ai_usage_log preserved
Read response.usage per call (Tier 1/2) and span.model_request_end events (Tier 3). Same Supabase sink + Sentry.
Redis SSE retained
Workflow streaming stays our layer. Chat moves to session event SSE; bridge events into the existing frontend SSE contract.
Caching + batches
Cache frozen system prompts (~90% input savings). Route batch scoring (auditor, qualifier) through the Batches API at 50%.
Phased rollout
executionLowest-risk first. Each phase ships behind a flag with shadow-comparison against the live Agno path before cutover.
Foundation · SDK client, model swap harness, eval baseline
Stand up the Anthropic client, port guardrails/validators, and capture a golden-output eval set from the live gpt-5.4 path so every later phase has a regression baseline.
- Anthropic SDK + auth in
ac-python-apisettings - Port
PromptInjectionGuardrail+ topic validators - Eval harness: capture N representative runs per agent
Tier 1 agents · ~26 single-shot extractors
Convert each Agno Agent to messages.parse(). Highest count, lowest risk - no control-flow change. Shadow-run against the baseline; gate on schema-match + quality eval.
- Start with demo generators + validators (non-customer-facing)
- Then strategist / auditor / qualifier / writers
- Enable prompt caching on frozen system prompts
Tier 2 workflows · Envoy → Headhunter → Sonar
Swap LLM nodes inside the existing DAGs, smallest first. Orchestration, streaming, telemetry untouched. Verify parallel fan-out + ordering parity per workflow.
- Envoy (3 steps) as the canary
- Headhunter (tool-runner node + Hunter enrich)
- Sonar last (7 steps, highest fan-out)
Tier 3 chat → Managed Agents · the genuine managed-agent
Create the persisted chat agent, port AcCliToolkit to custom tools, wire the SSE bridge + write-confirmation policy. Run side-by-side with the bespoke runner before cutover.
- Environment + agent provisioned via
antYAML in repo - Custom-tool host-side exec + allowlist
- Frontend SSE contract bridge + smoke test
Decommission Agno · remove dependency
Delete agno from pyproject.toml, drop agno_client_manager and BaseAgnoWorkflow, clean up wrappers once all 62 files are off the runtime.
Risks & open questions
unknownsOutput-quality drift on model swap
gpt-5.4 → opus-4-8 changes prompt sensitivity and structured-output behavior. Mitigation: eval baseline in Phase 1; shadow-compare every agent before cutover; re-tune prompts per the migration guide.
Managed Agents is beta
Chat (Tier 3) depends on a beta surface (event semantics, SSE reconnect, vault auth). Mitigation: keep the bespoke runner behind a flag; Tier 1/2 use GA API and carry the bulk of the migration.
SSE contract bridge for chat
Session event stream ≠ today’s Redis/frontend SSE shape. Mitigation: adapter layer translating agent.message / tool events into the existing frontend contract; covered by E2E smoke.
Cost profile shift
opus-4-8 token counting + effort differ from gpt-5.4. Mitigation: prompt caching, low effort on extractors, Batches API for scoring; re-baseline ai_usage_log dashboards.
Should any workflow step become model-driven?
e.g. Sonar website-enrichment fallback might benefit from a small Managed Agent sub-session. Decide in Phase 3 - default is to keep it deterministic unless evals show the agent loop wins.