要点
- Spotify のプラグイン shunt は、大きなファイルの読み込みと定型コード生成を安価な Worker モデルへ逃がします。bulk-read では平均約90%のトークン削減を報告しています。
- 減っているのは「総入力量」ではありません。ファイルの中身自体は Worker がそのまま読んでいます。減るのは「高価な Main Agent が読む量」です。
- この仕組みは依存が薄く、Python 標準ライブラリだけで再現できます。この記事では、Claude Code や Codex CLI のサブスクリプションを Worker として使う実装を紹介します。実際に動かしたログもあわせて示します。
- コード生成を Worker に任せる場合は、生成物を検証せずに使わないことが前提になります。
はじめに
対象読者は、Claude Code を使っていてコンテキストウィンドウやトークン消費に悩んでいる方です。前提として、Python 3.11 以上と、Claude Code か Codex CLI のいずれかのサブスクリプション契約が必要です。API キーは使いません。
大きなファイルを Main Agent に読ませると、それだけでコンテキストの大半を消費します。読んだ内容のうち実際に必要なのは要約の数行だけということも多く、残りは会話が続くほど圧迫要因になります。Spotify のエンジニアリングチームが公開しているプラグイン shunt は、この種の I/O 作業を安価なモデルへ逃がす設計を採っています。この記事では、shunt が何をしているかを確認したうえで、同じ考え方を Python と手元のサブスクリプションで再現します。
Spotify の shunt プラグインは何をしているか
shunt は Claude Code 向けのプラグインで、hooks scripts skills の3層構造になっています。
| 層 | 役割 |
|---|---|
| Hooks | 大きなファイルの読み込みをブロックし、bulk-reader スキルへ誘導します |
| Scripts | Worker モデルの呼び出しと出力の後処理を担当します |
| Skills | Claude にスクリプトをいつどう呼ぶかを伝えます |
Read フックの挙動は README に次のように書かれています。
Fires on every
Readtool call. Blocks full-file reads on files exceedingMIN_LINES(default: 350, configurable viaSHUNT_MIN_LINESenv var).
出典: spotify/portal-ai-plugins の shunt README
つまり、350行を超えるファイルの全文読み込みは既定でブロックされ、行数のしきい値は環境変数 SHUNT_MIN_LINES で変更できます。Bash 経由の cat head tail なども同様にフックで捕捉する設計です。
shunt は2つの Mode を持ちます。bulk-reader は複数ファイルを読んで質問に答え、要約だけを返します。code-writer は仕様と参照ファイルからコードを生成し、直接ディスクへ書き込みます。両モードとも temperature は 0.2 に設定されています。両モードの Worker には Google Gemini 2.5 Flash を使っています。これは Spotify Engineering ブログの記述によるものです。
162,000行規模の Java モノレポで計測したベンチマークは次のとおりです。
| シナリオ | 行数 | shunt なし | shunt あり | 削減率 |
|---|---|---|---|---|
| 単一の大きなファイル | 4,014 | 33,684 tokens | 5,737 tokens | 82% |
| ソースとテストのペア | 7,408 | 75,990 tokens | 4,148 tokens | 94% |
| 複数ファイル横断 | 1,281 | 16,221 tokens | 821 tokens | 94% |
| code-write | 3,667 | 40,614 tokens + 生成コスト | 833行をディスクへ書き込み | - |
bulk-read の平均削減率は 90% です。
一方で shunt は何でも委譲するわけではありません。デバッグ、正確な編集、350行未満の小さいファイル、アーキテクチャ判断は Main Agent に残す設計です。理由と具体的な内容は後述の「逃がすもの／残すもの」で扱います。
なぜ「I/O」なのか
先ほどの数値を思い出してください。たとえば「4,014行のファイルで 33,684 tokens が 5,737 tokens に減る」という例です。これだけ見ると、まるで入力自体が圧縮されたように見えます。実際に起きているのは違います。
通常の読み込みでは、ファイルの中身がそのまま Main Agent のコンテキストに入ります。
shunt を使う場合、ファイルの中身は変わらず全文が読まれます。ただし読むのは Worker であって、Main Agent ではありません。Main Agent が受け取るのは Worker の要約だけです。
総入力量そのものが 34,000 から 2,000 に減るわけではありません。減っているのは、高価な Main Agent が読む量です。ファイルの全文を読むという作業自体はなくなっておらず、それを担当するモデルが入れ替わっているだけです。
code-writer はさらに一歩進んでいます。生成したコードを Main Agent に一切返さず、直接ディスクへ書き込みます。Main Agent が受け取るのは「何行書いた」という受領書だけです。これにより、Worker の出力トークンまで高価な Main Agent の入力から除外できます。
サブスクリプションで動かす
ここからは、この考え方を Python で再現したサンプルを説明します。設計上の判断は主に3つです。
1つ目は、ルーティングの単位をモデルではなく Mode にしたことです。bulk-reader や code-writer という役割を先に決め、その役割にどのモデルを割り当てるかは別問題として扱います。こうしておけば、Worker のモデルを Gemini から別のモデルに変えても、ルーターのコードには手を入れずに済みます。
2つ目は、ルーティングの判定をヒューリスティックだけで行い、LLM に判断させないことです。行数が350を超えるか、対象ファイルが3つ以上かという条件は、コード側で決定的に判定できます。判定のためだけに別の LLM 呼び出しを挟むのは、それ自体がコストになります。
3つ目は、Worker を API キーではなくサブスクリプション CLI で動かすことです。API キーを別途契約すると、読者が試すたびに追加の課金が発生します。多くの方はすでに Claude Code や ChatGPT のサブスクリプションを契約しているはずです。その CLI を Worker として呼び出せば追加費用なしで試せます。
Claude CLI を Worker として呼ぶ場合、claude -p に --system-prompt を渡します。これはデフォルトのシステムプロンプトを完全に差し替えるオプションです。似た名前の --append-system-prompt とは挙動が異なります。--append-system-prompt で試したところ、Claude Code の既定のコーディングエージェント人格が残ってしまいました。その結果、JSON での分類結果ではなく「プロジェクトのパスを教えてください」のような対話的な応答が返ることが複数回ありました。この挙動は、同じ claude -p を分類器として使う別の実験で先に確認したもので、Worker にも同じ設計を採用しました。--system-prompt に切り替えてからは安定して意図した出力が返るようになりました。あわせて --allowedTools "" を指定しています。空の指定でツールが使われなくなることは、手元の実行で確認しました。
claude CLI が JSON で返す usage.input_tokens には、prompt cache から読んだ分が含まれません。458行のファイルを読ませても 10 程度と出て紛らわしい値でした。そのためサンプルでは、cache 関連の2項目を足した値を Worker input tokens として表示しています。それでも CLI 自身のシステムプロンプトなどの分が含まれるため、ファイル本体の概算より大きく出ます。Worker 側の数値は目安として扱ってください。
Codex CLI を Worker として呼ぶ場合は codex exec --json を使います。標準出力に流れる JSONL イベントを1行ずつパースします。item.completed で type が agent_message の行だけを拾って本文を取り出します。turn.completed イベントからは使用トークン数を取得します。無関係な MCP サーバーの認証エラーなどもイベントとして流れてくるため、拾うイベント種別を絞り込むことが必要でした。プロンプトはファイル全文を含んで長くなるため、コマンドライン引数ではなく標準入力で渡しています。引数で渡すと OS の引数長の上限に当たることがあります。
サンプルコード全文
設計のポイントを実装で確認します。まず Mode の定義です。Mode は名前と Worker への指示文だけを持つ単純なデータクラスにしています。
@dataclass(frozen=True)
class Mode:
name: str
instructions: str
BULK_READER = Mode(
name="bulk-reader",
instructions="""You are a precise code analyst.
Read the provided files and answer the question concisely.
Rules:
- Output structured bullets only.
- No greetings or preambles.
- Lead bullets with exact class, function, file, or symbol names.
- Only discuss information relevant to the question.
- Do not make unsupported assumptions.
- Do not ask clarifying questions. Do not investigate further files.
Answer directly using only the given content.""",
)
Worker の呼び出し口は抽象クラスにして、CLI 側の実装から切り離しています。ClaudeCliWorker が --system-prompt と --allowedTools "" を使っている箇所です。
class WorkerBackend(abc.ABC):
"""Worker LLMの呼び出し口。実装ごとにサブスクCLI/APIへ差し替える。"""
@abc.abstractmethod
def invoke(self, instructions: str, message: str) -> LLMResult:
raise NotImplementedError
class ClaudeCliWorker(WorkerBackend):
"""claude CLI(Claude Code Pro/Max契約)をWorkerとして使う。"""
TIMEOUT_SECONDS = 300
MAX_ATTEMPTS = 2 # 初回 + 1回だけ自動リトライ
def __init__(self) -> None:
self.model = os.getenv("SHUNT_WORKER_MODEL", "haiku")
def invoke(self, instructions: str, message: str) -> LLMResult:
last_error = "unknown error"
for attempt in range(1, self.MAX_ATTEMPTS + 1):
try:
return self._invoke_once(instructions, message)
except subprocess.TimeoutExpired:
last_error = f"{self.TIMEOUT_SECONDS}秒でタイムアウトしました"
except RuntimeError as exc:
last_error = str(exc)
print(f"[ClaudeCliWorker] 試行{attempt}/{self.MAX_ATTEMPTS} 失敗: {last_error}", file=sys.stderr)
# 生トレースバックではなく1行メッセージで終了する
print("Worker がタイムアウトしました。もう一度実行するか SHUNT_WORKER_MODEL を変えてください。", file=sys.stderr)
raise SystemExit(1)
ルーターは Spotify の check-file-size フックと同じ考え方を、決定的なロジックとして実装しています。ファイルが3つ以上、またはいずれかが350行を超える場合に bulk-reader へ振り分けます。
class ShuntRouter:
def __init__(self, min_lines: int = 350, multi_file_threshold: int = 3) -> None:
self.min_lines = min_lines
self.multi_file_threshold = multi_file_threshold
def route_read(self, paths: list[Path]) -> RouteDecision:
if len(paths) >= self.multi_file_threshold:
return RouteDecision(
mode="bulk-reader",
reason=f"{len(paths)} files >= {self.multi_file_threshold}",
)
for path in paths:
lines = count_lines(path)
if lines > self.min_lines:
return RouteDecision(
mode="bulk-reader",
reason=f"{path} has {lines} lines > threshold {self.min_lines}",
)
return RouteDecision(mode="main-agent", reason="Small targeted read")
CodeWriter の書き込み部分です。Worker が生成したコードを Main Agent へ返さず、そのままディスクへ書いています。
def execute(self, spec: str, reference: Path, target: Path, overwrite: bool = False) -> CodeWriteResult:
if target.exists() and not overwrite:
print(f"{target} は既に存在します。上書きするには --overwrite を付けてください。", file=sys.stderr)
raise SystemExit(1)
reference_code = reference.read_text(encoding="utf-8", errors="replace")
message = (
f"Specification:\n{spec}\n\n"
f'Reference file:\n<file path="{reference}">\n{reference_code}\n</file>'
)
code = ""
result: Optional[LLMResult] = None
valid = True
for attempt in range(1, self.MAX_GENERATION_ATTEMPTS + 1):
result = self.worker.invoke(CODE_WRITER.instructions, message)
code = extract_code(result.text)
valid = target.suffix != ".py" or self._is_valid_python(code, target)
if valid:
break
print(f"[CodeWriter] 試行{attempt}/{self.MAX_GENERATION_ATTEMPTS}: 生成結果が有効なPythonコードではありませんでした", file=sys.stderr)
message += self.RETRY_HINT
if not valid:
print("生成結果がコードとして解釈できませんでした(targetは書き込んでいません)", file=sys.stderr)
raise SystemExit(1)
# 重要: 生成コードはMain Agentへ返さずディスクへ直接書く。Workerの出力を
# 検証せずに書き込む(.py以外は構文検証もしない)ため、targetは慎重に選ぶこと。
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(code + "\n", encoding="utf-8")
return CodeWriteResult(
target=target,
generated_lines=len(code.splitlines()),
worker_input_tokens=result.input_tokens if result else None,
worker_output_tokens=result.output_tokens if result else None,
)
全文は次のとおりです（約400行）。SHUNT_BACKEND 環境変数で claude codex openai の3つの Worker 実装を切り替えられます。
shunt.py 全文（約400行）
"""Spotify Shunt風の2 Mode(bulk-reader / code-writer)ルーター、サブスク版。
依存ゼロ(httpxすら使わない)。Worker LLMは SHUNT_BACKEND=claude(既定・claude
CLI)|codex(codex CLI)|openai(LLM_BASE_URL/LLM_API_KEY)のいずれかへ委譲する。
python3 shunt.py bulk-read --question "..." --paths a.py b.py
python3 shunt.py code-write --spec "..." --reference ref.py --target out.py
"""
from __future__ import annotations
import argparse
import abc
import json
import math
import os
import re
import subprocess
import sys
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Optional
# ---- Mode definition ----
@dataclass(frozen=True)
class Mode:
name: str
instructions: str
BULK_READER = Mode(
name="bulk-reader",
instructions="""You are a precise code analyst.
Read the provided files and answer the question concisely.
Rules:
- Output structured bullets only.
- No greetings or preambles.
- Lead bullets with exact class, function, file, or symbol names.
- Only discuss information relevant to the question.
- Do not make unsupported assumptions.
- Do not ask clarifying questions. Do not investigate further files.
Answer directly using only the given content.""",
)
CODE_WRITER = Mode(
name="code-writer",
instructions="""You generate code files based on a specification and reference file.
Rules:
- Match existing patterns, conventions, naming, and style.
- Output only code.
- Do not output explanations.
- Do not use Markdown fences.
- If ambiguous, make reasonable choices based on the reference code.
- Do not ask clarifying questions. Output the code directly.
- Do not add trailers, sign-offs, or attribution lines such as Co-Authored-By.
- Start your response with the first line of code. Do not write any sentence,
greeting, or summary before or after the code.""",
)
# ---- Worker backend abstraction ----
@dataclass
class LLMResult:
text: str
input_tokens: Optional[int] = None
output_tokens: Optional[int] = None
class WorkerBackend(abc.ABC):
"""Worker LLMの呼び出し口。実装ごとにサブスクCLI/APIへ差し替える。"""
@abc.abstractmethod
def invoke(self, instructions: str, message: str) -> LLMResult:
raise NotImplementedError
class ClaudeCliWorker(WorkerBackend):
"""claude CLI(Claude Code Pro/Max契約)をWorkerとして使う。"""
TIMEOUT_SECONDS = 300
MAX_ATTEMPTS = 2 # 初回 + 1回だけ自動リトライ
def __init__(self) -> None:
self.model = os.getenv("SHUNT_WORKER_MODEL", "haiku")
def invoke(self, instructions: str, message: str) -> LLMResult:
last_error = "unknown error"
for attempt in range(1, self.MAX_ATTEMPTS + 1):
try:
return self._invoke_once(instructions, message)
except subprocess.TimeoutExpired:
last_error = f"{self.TIMEOUT_SECONDS}秒でタイムアウトしました"
except RuntimeError as exc:
last_error = str(exc)
print(f"[ClaudeCliWorker] 試行{attempt}/{self.MAX_ATTEMPTS} 失敗: {last_error}", file=sys.stderr)
# 生トレースバックではなく1行メッセージで終了する
print("Worker がタイムアウトしました。もう一度実行するか SHUNT_WORKER_MODEL を変えてください。", file=sys.stderr)
raise SystemExit(1)
def _invoke_once(self, instructions: str, message: str) -> LLMResult:
proc = subprocess.run(
[
"claude", "-p", "--model", self.model, "--output-format", "json",
"--allowedTools", "", # Workerにツールは使わせない
"--system-prompt", instructions,
],
input=message, capture_output=True, text=True,
timeout=self.TIMEOUT_SECONDS, check=False,
)
if proc.returncode != 0:
raise RuntimeError(f"claude CLI failed (exit={proc.returncode}): {proc.stderr[:300]}")
data = json.loads(proc.stdout)
if data.get("is_error"):
raise RuntimeError(f"claude CLI error: {data.get('result')}")
usage = data.get("usage", {})
# input_tokens単体だとprompt cache分が抜けて過小表示になるため合算する
input_tokens = (
usage.get("input_tokens", 0)
+ usage.get("cache_read_input_tokens", 0)
+ usage.get("cache_creation_input_tokens", 0)
)
return LLMResult(
text=data["result"],
input_tokens=input_tokens,
output_tokens=usage.get("output_tokens"),
)
class CodexCliWorker(WorkerBackend):
"""codex CLI(ChatGPT契約)をWorkerとして使う。"""
def __init__(self) -> None:
self.model = os.getenv("SHUNT_WORKER_MODEL", "gpt-5.6-luna")
def invoke(self, instructions: str, message: str) -> LLMResult:
prompt = f"System instructions:\n{instructions}\n\n{message}"
# 位置引数でpromptを渡すとARG_MAX超過の恐れがあるため、stdin経由で渡す
proc = subprocess.run(
["codex", "exec", "--skip-git-repo-check", "--json", "-m", self.model],
input=prompt, capture_output=True, text=True, timeout=180, check=False,
)
if proc.returncode != 0:
print(f"codex CLI failed (exit={proc.returncode}): {proc.stderr[-300:]}", file=sys.stderr)
raise SystemExit(1)
text: Optional[str] = None
usage: dict = {}
for line in proc.stdout.splitlines():
line = line.strip()
if not line.startswith("{"):
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
item = event.get("item", {})
if event.get("type") == "item.completed" and item.get("type") == "agent_message":
text = item.get("text")
elif event.get("type") == "turn.completed":
usage = event.get("usage", {})
if text is None:
raise RuntimeError(f"codex CLI から agent_message を取得できませんでした: {proc.stdout[-500:]}")
return LLMResult(
text=text,
input_tokens=usage.get("input_tokens"),
output_tokens=usage.get("output_tokens"),
)
class OpenAICompatWorker(WorkerBackend):
"""OpenAI互換Chat Completions API(APIキー方式)。urllib.requestのみで実装。"""
def __init__(self) -> None:
base_url = os.environ.get("LLM_BASE_URL")
api_key = os.environ.get("LLM_API_KEY")
if not base_url or not api_key:
print("SHUNT_BACKEND=openai には LLM_BASE_URL と LLM_API_KEY が必要です", file=sys.stderr)
raise SystemExit(1)
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.model = os.getenv("SHUNT_WORKER_MODEL", "gpt-5.6-luna") # gatewayで使えるモデル名に置き換える
def invoke(self, instructions: str, message: str) -> LLMResult:
payload = json.dumps({
"model": self.model,
"temperature": 0.2,
"messages": [
{"role": "system", "content": instructions},
{"role": "user", "content": message},
],
}).encode("utf-8")
request = urllib.request.Request(
f"{self.base_url}/chat/completions",
data=payload,
headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=180) as response:
data = json.loads(response.read())
usage = data.get("usage", {})
return LLMResult(
text=data["choices"][0]["message"]["content"],
input_tokens=usage.get("prompt_tokens"),
output_tokens=usage.get("completion_tokens"),
)
def get_backend() -> WorkerBackend:
backend = os.getenv("SHUNT_BACKEND", "claude").lower()
if backend == "claude":
return ClaudeCliWorker()
if backend == "codex":
return CodexCliWorker()
if backend == "openai":
return OpenAICompatWorker()
raise ValueError(f"unknown SHUNT_BACKEND: {backend!r} (claude|codex|openai)")
# ---- Utilities ----
def approximate_tokens(text: str) -> int:
"""粗い概算。本番ではproviderのtokenizerに置き換える。"""
return math.ceil(len(text) / 4)
def count_lines(path: Path) -> int:
with path.open("r", encoding="utf-8", errors="replace") as f:
return sum(1 for _ in f)
_TRAILER_RE = re.compile(r"^(Co-Authored-By|Signed-off-by):", re.IGNORECASE)
def extract_code(text: str) -> str:
"""先頭に確認文が付いていても、テキスト全体から最初のフェンスブロックの
中身だけを取り出す(フェンスが無ければ全文を使う)。末尾のtrailer行・
空行は除去する。"""
lines = text.strip("\n").splitlines()
start = next((i for i, line in enumerate(lines) if line.strip().startswith("```")), None)
if start is not None:
end = next(
(j for j in range(start + 1, len(lines)) if lines[j].strip().startswith("```")),
len(lines),
)
lines = lines[start + 1 : end]
# 末尾のtrailer行・空行を除去
while lines and (not lines[-1].strip() or _TRAILER_RE.match(lines[-1].strip())):
lines.pop()
return "\n".join(lines).strip()
# ---- Mode 1: bulk-reader ----
@dataclass
class BulkReadResult:
answer: str
baseline_main_tokens: int
delegated_main_tokens: int
main_token_saving_pct: float
worker_input_tokens: Optional[int]
worker_output_tokens: Optional[int]
class BulkReader:
def __init__(self, worker: WorkerBackend) -> None:
self.worker = worker
def execute(self, paths: Iterable[Path], question: str) -> BulkReadResult:
parts: list[str] = []
original_text = ""
for path in paths:
content = path.read_text(encoding="utf-8", errors="replace")
original_text += content
parts.append(f'<file path="{path}">\n{content}\n</file>')
message = "\n\n".join(parts) + f"\n\nQuestion:\n{question}"
result = self.worker.invoke(BULK_READER.instructions, message)
baseline = approximate_tokens(original_text)
delegated = approximate_tokens(result.text)
saving = (1 - delegated / baseline) * 100 if baseline else 0.0
return BulkReadResult(
answer=result.text,
baseline_main_tokens=baseline,
delegated_main_tokens=delegated,
main_token_saving_pct=saving,
worker_input_tokens=result.input_tokens,
worker_output_tokens=result.output_tokens,
)
# ---- Mode 2: code-writer ----
@dataclass
class CodeWriteResult:
target: Path
generated_lines: int
worker_input_tokens: Optional[int]
worker_output_tokens: Optional[int]
class CodeWriter:
MAX_GENERATION_ATTEMPTS = 2 # 初回 + 構文エラー時に1回だけ書き直しを依頼
RETRY_HINT = (
"\n\nYour previous answer was not valid code. Return only the file "
"content, starting with the first line of code."
)
def __init__(self, worker: WorkerBackend) -> None:
self.worker = worker
def execute(self, spec: str, reference: Path, target: Path, overwrite: bool = False) -> CodeWriteResult:
if target.exists() and not overwrite:
print(f"{target} は既に存在します。上書きするには --overwrite を付けてください。", file=sys.stderr)
raise SystemExit(1)
reference_code = reference.read_text(encoding="utf-8", errors="replace")
message = (
f"Specification:\n{spec}\n\n"
f'Reference file:\n<file path="{reference}">\n{reference_code}\n</file>'
)
code = ""
result: Optional[LLMResult] = None
valid = True
for attempt in range(1, self.MAX_GENERATION_ATTEMPTS + 1):
result = self.worker.invoke(CODE_WRITER.instructions, message)
code = extract_code(result.text)
valid = target.suffix != ".py" or self._is_valid_python(code, target)
if valid:
break
print(f"[CodeWriter] 試行{attempt}/{self.MAX_GENERATION_ATTEMPTS}: 生成結果が有効なPythonコードではありませんでした", file=sys.stderr)
message += self.RETRY_HINT
if not valid:
print("生成結果がコードとして解釈できませんでした(targetは書き込んでいません)", file=sys.stderr)
raise SystemExit(1)
# 重要: 生成コードはMain Agentへ返さずディスクへ直接書く。Workerの出力を
# 検証せずに書き込む(.py以外は構文検証もしない)ため、targetは慎重に選ぶこと。
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(code + "\n", encoding="utf-8")
return CodeWriteResult(
target=target,
generated_lines=len(code.splitlines()),
worker_input_tokens=result.input_tokens if result else None,
worker_output_tokens=result.output_tokens if result else None,
)
@staticmethod
def _is_valid_python(code: str, target: Path) -> bool:
try:
compile(code, str(target), "exec")
except SyntaxError:
return False
return True
# ---- Router (決定的。LLMに判断させない) ----
@dataclass
class RouteDecision:
mode: str
reason: str
class ShuntRouter:
def __init__(self, min_lines: int = 350, multi_file_threshold: int = 3) -> None:
self.min_lines = min_lines
self.multi_file_threshold = multi_file_threshold
def route_read(self, paths: list[Path]) -> RouteDecision:
if len(paths) >= self.multi_file_threshold:
return RouteDecision(
mode="bulk-reader",
reason=f"{len(paths)} files >= {self.multi_file_threshold}",
)
for path in paths:
lines = count_lines(path)
if lines > self.min_lines:
return RouteDecision(
mode="bulk-reader",
reason=f"{path} has {lines} lines > threshold {self.min_lines}",
)
return RouteDecision(mode="main-agent", reason="Small targeted read")
# ---- CLI ----
def _cmd_bulk_read(args: argparse.Namespace) -> None:
paths = [Path(p) for p in args.paths]
decision = ShuntRouter().route_read(paths)
print(f"[route] {decision.mode}: {decision.reason}")
if decision.mode != "bulk-reader":
print("Small read: let Main Agent read the files directly.")
return
worker = BulkReader(get_backend())
result = worker.execute(paths, args.question)
print("\n--- Worker answer ---")
print(result.answer)
print("\n--- Token comparison ---")
print("Baseline Main Agent tokens:", result.baseline_main_tokens)
print("Delegated Main Agent tokens:", result.delegated_main_tokens)
print("Main Agent saving:", f"{result.main_token_saving_pct:.1f}%")
print("Worker input tokens:", result.worker_input_tokens)
print("Worker output tokens:", result.worker_output_tokens)
def _cmd_code_write(args: argparse.Namespace) -> None:
worker = CodeWriter(get_backend())
result = worker.execute(
spec=args.spec,
reference=Path(args.reference),
target=Path(args.target),
overwrite=args.overwrite,
)
# Main Agentにはこの一行の「受領書」だけ返す
print(f"Wrote {result.generated_lines} lines to {result.target}")
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command", required=True)
bulk = sub.add_parser("bulk-read")
bulk.add_argument("--question", required=True)
bulk.add_argument("--paths", nargs="+", required=True)
bulk.set_defaults(func=_cmd_bulk_read)
writer = sub.add_parser("code-write")
writer.add_argument("--spec", required=True)
writer.add_argument("--reference", required=True)
writer.add_argument("--target", required=True)
writer.add_argument("--overwrite", action="store_true", help="targetが既存でも上書きする")
writer.set_defaults(func=_cmd_code_write)
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()
使い方
前提は Python 3.11以上です。claude（Claude Code Pro/Max契約）または codex（ChatGPT契約）へのログインが必要です。API キーは SHUNT_BACKEND=openai を使う場合だけ必要になります。
環境変数は次のとおりです。
| 変数 | 既定値 | 説明 |
|---|---|---|
SHUNT_BACKEND |
claude |
claude | codex | openai
|
SHUNT_WORKER_MODEL |
backendごとに haiku / gpt-5.6-luna / gpt-5.6-luna
|
Workerに使うモデル |
LLM_BASE_URL / LLM_API_KEY
|
- |
SHUNT_BACKEND=openai のときのみ必須 |
# 350行超のファイルはbulk-readerへ自動委譲、要約だけをMain Agentが受け取る
python3 shunt.py bulk-read --question "認証処理の流れを説明してください" \
--paths fixtures/auth_service.py
# 参照ファイルのパターンを真似てテストを生成し、直接ディスクへ書く
python3 shunt.py code-write --spec "AuthService の unit test を作成してください" \
--reference fixtures/order_service_test.py --target out/auth_service_test.py
# 350行以下の小さいファイルはMain Agentが直接読む(ルーティングされない)
python3 shunt.py bulk-read --question "何をする関数か" --paths fixtures/small_util.py
# targetが既に存在する場合は --overwrite を付けない限りエラー終了する
python3 shunt.py code-write --spec "..." --reference ref.py --target out.py --overwrite
手元の検証では、fixtures として 458行の架空の認証モジュール（auth_service.py）を用意しました。参照ファイルには order_service_test.py（72行）という小さなテストを使いました。中身は検証専用の架空コードなので全文は載せません。試すときは、お使いのリポジトリの中で実際に350行を超えているファイルを指定してください。上の3コマンドがそのまま使えます。
動かしてみる
実際に動かしたログを、実行結果そのままで示します。環境は macOS、Python 3.13.0 です。Worker には claude CLI 2.1.273（サブスクリプション認証済み）を使いました。もう一方の Worker には codex CLI 0.154.0（ChatGPT認証済み）を使いました。
bulk-read（claude backend）
458行の auth_service.py を対象に実行すると、ルーターは自動で bulk-reader へ振り分けました。
[route] bulk-reader: fixtures/auth_service.py has 458 lines > threshold 350
--- Token comparison ---
Baseline Main Agent tokens: 3614
Delegated Main Agent tokens: 500
Main Agent saving: 86.2%
Worker input tokens: 51863
Worker output tokens: 1898
Worker の回答は認証フローを箇条書きで整理したもので、Main Agent が読むトークン数は 3,614 から 500 まで減りました。Worker input tokens の 51,863 には、ファイル本体に加えて CLI 側のシステムプロンプトなどの分も含まれています。
code-write（claude backend）
order_service_test.py を参照ファイルとして、AuthService のユニットテストを生成させました。
Wrote 137 lines to out/r2_auth_service_test_run1.py
Wrote 174 lines to out/r2_auth_service_test_run2.py
Wrote 254 lines to out/r2_auth_service_test_run3.py
Wrote 131 lines to out/r2_auth_service_test_run4.py
Wrote 121 lines to out/r2_auth_service_test_run5.py
Main Agent が受け取るのは、この1行ずつだけです。安定性を見るため、既定の Worker（haiku）で同じ指示を5回連続実行しました。5回とも生成ファイルは python3 -m py_compile を通りました。一方で python3 -m unittest の結果は次のとおり、3回が green、2回が赤でした。
| 実行 | 生成行数 | py_compile | unittest |
|---|---|---|---|
| 1 | 137 | OK | 11 tests, OK |
| 2 | 174 | OK | 失敗（アサーションの内容が誤り） |
| 3 | 254 | OK | 失敗（同上） |
| 4 | 131 | OK | 9 tests, OK |
| 5 | 121 | OK | 9 tests, OK |
赤になった2回は、構文としては正しいものの、テストの中身が実装の仕様と食い違っていました。たとえば UUID が入る項目に固定の文字列を期待する、といった誤りです。これは安価な Worker の限界で、Spotify がテスト雛形の生成を Worker に任せつつ「生成物は検証する」としているのと同じ話です。
次に Worker を SHUNT_WORKER_MODEL=sonnet に切り替え、同じ指示を5回実行しました。こちらは5回とも py_compile と unittest が通りました。
Wrote 108 lines to out/sonnet_auth_service_test_run1.py
Wrote 100 lines to out/sonnet_auth_service_test_run2.py
[CodeWriter] 試行1/2: 生成結果が有効なPythonコードではありませんでした
Wrote 100 lines to out/sonnet_auth_service_test_run3.py
Wrote 116 lines to out/sonnet_auth_service_test_run4.py
Wrote 123 lines to out/sonnet_auth_service_test_run5.py
| 実行 | 生成行数 | unittest | 自動再試行 |
|---|---|---|---|
| 1 | 108 | 7 tests, OK | なし |
| 2 | 100 | 8 tests, OK | なし |
| 3 | 100 | 8 tests, OK | 1回 |
| 4 | 116 | 9 tests, OK | なし |
| 5 | 123 | 9 tests, OK | なし |
3回目は Worker の1回目の応答がコードとして解釈できず、サンプルが自動で書き直しを依頼して2回目で成功しています。私の手元では、bulk-reader は haiku、code-writer は sonnet という使い分けが落としどころでした。
サンプルがこの形になるまでには、いくつかの失敗がありました。生成コードの末尾に Markdown のフェンスや Co-Authored-By の署名行が残る回がありました。コードの前に日本語の説明文が付く回や、コードを返さず完成報告の散文だけが返る回もありました。最後のケースでは、散文をそのままファイルに書いて「Wrote 11 lines」と成功表示してしまいました。そこでサンプルでは、フェンスをテキスト全体から探して中身だけを取り出し、.py への書き込み前に compile() で構文を検証しています。通らなければ1回だけ書き直しを依頼し、それでも通らなければファイルを書かずに終了します。
小さいファイル → main-agent ルート
28行の small_util.py を指定すると、ルーターは委譲せずに Main Agent へ読ませる判断を返しました。
[route] main-agent: Small targeted read
Small read: let Main Agent read the files directly.
閾値以下のファイルまで Worker に回すと、委譲のオーバーヘッドの方が大きくなります。ルーターがこの判断をコード側で決定的に行っている点が、ここで確認できます。
bulk-read（codex backend）
同じ auth_service.py を SHUNT_BACKEND=codex で実行した結果です。Worker は既定の gpt-5.6-luna です。Codex CLI のモデル一覧で「Fast and affordable」と説明されている、現行の安価なモデルです。
[route] bulk-reader: fixtures/auth_service.py has 458 lines > threshold 350
--- Token comparison ---
Baseline Main Agent tokens: 3614
Delegated Main Agent tokens: 250
Main Agent saving: 93.1%
Worker input tokens: 43540
Worker output tokens: 433
codex backend の Worker input tokens は 43,540 です。claude backend の 51,863 とは開きがあります。CLI ごとにシステムプロンプトやツール定義の量が違うため、Worker 側の入力量は横並びで比べられません。比べるなら Main Agent 側の削減率を見るのが確実です。
逃がすもの／残すもの
shunt の README には、Worker へ逃がすべきではない作業が明記されています。
| Cheap Worker に逃がす | Main / Frontier に残す |
|---|---|
| 大量ファイルの理解 | デバッグ |
| 複数ファイルの要約 | アーキテクチャ判断 |
| テスト雛形生成 | 複雑な推論 |
| config生成 | セキュリティ判断 |
| type stub | 正確な編集 |
| 定型documentation | 曖昧な要件からの設計 |
デバッグやアーキテクチャ判断は Claude の推論そのものが必要な作業であり、要約に変換すると情報が失われます。編集についても、Claude が正確な内容をコンテキストに持っている必要があります。そのため offset や limit を指定した狙い撃ちの読み込みに留めるべきだとされています。
code-writer は Worker の出力を、構文の検証だけを通してディスクへ書き込みます。構文が通ることと内容が正しいことは別なので、生成されたコードは必ずテストで確認する必要があります。既存ファイルがある場合は --overwrite を付けない限り書き込まないようにしています。
Hook で強制するなら
ここまでのサンプルはコマンドを手動で呼ぶ形でした。shunt はさらに一歩進めて、Read ツールそのものをフックでブロックしています。
check-file-size フックのロジックを言葉で整理すると次のようになります。offset や limit を指定した狙い撃ちの読み込みは許可します。ファイルが350行以下なら許可します。それ以外、つまり350行を超えるファイルの全文読み込みは block し、bulk-reader への切り替えを促します。
Claude Code には PreToolUse というフックの仕組みがあります。Read ツール呼び出しの前に任意のスクリプトを挟んで許可・拒否を判定できます。この仕組みを使えば、check-file-size と同じ判定を自分の環境にも組み込めます。設定方法は公式ドキュメントの Hooks のページにまとまっています。
まとめ
shunt の本質は、モデルを直接選ぶことではなく、I/O の多い作業を専用の Mode へ切り出すことにあります。bulk-reader はファイルの全文読み込みを Worker へ回して要約だけを返します。code-writer は生成したコードを Main Agent に返さず、直接ディスクへ書きます。総入力量が減るのではなく、高価な Main Agent が読む量が減るという点が、この設計の要です。
サブスクリプション CLI を Worker として使えば、API キーの追加契約なしに同じ考え方を試せます。決定的なルーティングと、Worker が担当しない領域（デバッグ、編集、アーキテクチャ判断）を明確に分けておくことが、実装を安全に保つ鍵になります。
参考
- Spotify Engineering ブログ「Portal by Spotify cut my Claude Code token usage by 90%」 https://engineering.atspotify.com/2026/9/portal-by-spotify-cut-my-claude-code-token-usage-by-90
- spotify/portal-ai-plugins の shunt プラグイン README https://github.com/spotify/portal-ai-plugins/tree/main/plugins/shunt
- Claude Code 公式ドキュメント「Run Claude Code programmatically」 https://code.claude.com/docs/en/headless
- Claude Code 公式ドキュメント「Hooks」 https://code.claude.com/docs/en/hooks
- 関連記事: トークン節約 ── 最適な LLM ルーティングについて考えてみた https://qiita.com/nogataka/items/ff1ac78c7dafbe42369b