The agent core and prompt strategies

claude_skills_agent.py is the host-agnostic core. Both surfaces use it unchanged. It holds four module constants, all overridable at deploy time without a rebuild:

create_agent() runs once at startup. It confirms claude-agent-sdk imports, raises if neither ANTHROPIC_API_KEY nor CLAUDE_CODE_USE_FOUNDRY=1 is set, and logs a warning if SKILL.md is missing from <cwd>/.claude/skills/<skill-name>/. In that case the agent still starts, but skill loading fails.

The SDK options

base_tools = ["Bash", "Read", "Write", "Glob", "Grep", "Skill"]
kwargs: Dict[str, Any] = dict(
    model=DEFAULT_MODEL,
    cwd=SKILL_PARENT_DIR,                  # /app, where .claude/skills/ lives
    tools=base_tools,                      # the tools DEFINED in the request
    allowed_tools=base_tools,              # the tools PERMITTED to run
    setting_sources=["project"],           # load skills from <cwd>/.claude/skills/
    system_prompt={"type": "preset", "preset": "claude_code"},
    permission_mode="dontAsk",             # deny outside the allowlist, never prompt
    max_turns=MAX_TURNS,
    env={
        "DISABLE_AUTOUPDATER": "1",
        "CLAUDE_CODE_MAX_RETRIES": os.getenv("CLAUDE_CODE_MAX_RETRIES", "5"),
        "REPORT_OUTPUTS_DIR": OUTPUTS_DIR,  # from {{SKILL_ENV_VARS}}
    },
)
budget = (os.getenv("CLAUDE_AGENT_MAX_BUDGET_USD") or "").strip()
if budget:
    kwargs["max_budget_usd"] = float(budget)
effort = (os.getenv("CLAUDE_AGENT_EFFORT") or "").strip().lower()
if effort in ("low", "medium", "high", "xhigh", "max"):
    kwargs["effort"] = effort
buffer_bytes = (os.getenv("CLAUDE_AGENT_MAX_BUFFER_BYTES") or "").strip()
if buffer_bytes:
    kwargs["max_buffer_size"] = int(buffer_bytes)
options = ClaudeAgentOptions(**kwargs)

Event translation

For the plain-HTTP surface, run_conversation_stream() yields strings with a small wire format, which __main__.py maps to NDJSON events:

• •AssistantMessage.error is checked on every assistant message. It carries a category such as authentication_failed, billing_error, rate_limit, invalid_request, server_error, or unknown. Without this check an auth or billing failure ends the run later with the vaguer produced no text response.
• •A run that yields no text ends with Error: agent completed but produced no text response.
• •When ResultMessage.is_error is set, the core logs its subtype and result so the real reason is visible.
• •When a run uses 90% or more of MAX_TURNS, the core logs an approaching cap warning.

Reading tokens and cost

ResultMessage has no flat input_tokens or output_tokens attributes. Reading them with getattr(msg, "input_tokens", 0) returns 0 on every run, silently. Tokens live in model_usage (per model, camelCase, covers the whole tree including sub-agents), with usage (snake_case, top-level loop only) as the fallback. Input is the sum of uncached, cache-read, and cache-creation tokens: inputTokens alone is only the small uncached remainder.

model_usage = getattr(msg, "model_usage", None)
if isinstance(model_usage, dict) and model_usage:
    inp = out = 0
    for entry in model_usage.values():
        if not isinstance(entry, dict):
            continue
        inp += (int(entry.get("inputTokens") or 0)
                + int(entry.get("cacheReadInputTokens") or 0)
                + int(entry.get("cacheCreationInputTokens") or 0))
        out += int(entry.get("outputTokens") or 0)
    if inp or out:
        return inp, out
# fallback: msg.usage with input_tokens, cache_read_input_tokens,
# cache_creation_input_tokens, output_tokens (excludes sub-agents)

The core stores the result on the instance: last_token_usage (prompt_tokens, completion_tokens, total_tokens), last_cost_usd from total_cost_usd, and last_num_turns.

Progress labels

_describe_tool() turns each tool call into a short label. Replace {{TOOL_SCRIPT_LABELS}} with a map from the skill's script names to labels. Unmapped Bash calls fall back to Bash: <first line of the command>; Skill calls read Loading skill: <name>; file tools read <Tool>: <path or pattern>.

if tool_name == "Bash":
    cmd = (tool_input or {}).get("command", "") or ""
    for script, label in (
        ("fetch_sources.py", "Fetching sources"),
        ("render.py", "Rendering output"),
    ):
        if script in cmd:
            return label

Prompt strategies

_build_prompt() decides how every incoming message reaches the SDK. Replace {{PROMPT_STRATEGY_BODY}} with one of two patterns, matching the trigger mode you chose.

Pattern A: single-purpose

Use it when the agent does one thing and every run should run the skill end to end. Even an empty message or a bare go triggers the full pipeline.

def _build_prompt(self, user_message: str) -> str:
    directive = (
        f"You are a single-purpose agent. Use the **{SKILL_NAME}** skill "
        f"to handle the user's request. Read its SKILL.md and run the "
        f"full pipeline. Always run the full pipeline; do not ask the "
        f"user to clarify unless a required argument is genuinely "
        f"ambiguous.\n\nUser request:\n"
    )
    return directive + (user_message or "").strip()

Pattern B: multi-purpose

Use it when the skill has several command surfaces or the request varies from run to run. Claude invokes the skill only when the request matches it, and can decline off-topic requests.

def _build_prompt(self, user_message: str) -> str:
    hint = (
        f"You have access to the {SKILL_NAME} skill. Use it for "
        f"reporting requests. Always pass --format html.\n\n"
    )
    return hint + (user_message or "").strip()

Whichever pattern you pick, confirm that SKILL_NAME matches the skill's frontmatter name: exactly, that allowed_tools includes "Skill", and that setting_sources=["project"] is set.