Curious builder0 XP earned · 300 to level 2

0 daysFinish a lesson to begin

Badge collection0 of 6 unlocked

46 small wins to finish your pathNext lesson →

Tool search for large tool sets

Tool search is the mechanism that keeps tool definitions out of the context window until Claude searches for them. The agent is given a summary of what is available and loads up to five definitions per search.

Last updated: 20 Oct, 2026 · Claude Agent SDK

Every lesson so far had one or two tools. Connect two real MCP servers and you can have sixty. The tool search page names the two things that then go wrong: fifty tool definitions can take 10,000 to 20,000 tokens of the context window on every turn, and tool selection accuracy falls off beyond thirty to fifty tools loaded at once.

A depot with ten more tools

One catalogue is enough to see the effect. These ten tools are built from a list of rows rather than ten hand-written functions, which is also how a generated catalogue usually arrives:

python

from claude_agent_sdk import create_sdk_mcp_server, tool

# ten tools, as a catalogue usually arrives: a name and a description per row
ROWS = [("stock_level", "How many of an order's part are in stock."),
        ("pack_size", "How many units ship in one box."),
        ("shelf_slot", "Which shelf an order is picked from."),
        ("courier_name", "Which courier carries an order."),
        ("label_weight", "The weight printed on an order's label."),
        ("customs_code", "The customs code for an order."),
        ("return_window", "How many days are left to return an order."),
        ("insurance_band", "The insurance band an order falls in."),
        ("pallet_height", "The height of the pallet an order travels on."),
        ("fragile_flag", "Whether an order is marked fragile.")]

python

def depot_tool(name, description):
    """One catalogue tool, so ten of them take ten rows rather than ten functions."""
    @tool(name, description, {"order_id": str})
    async def handler(args, _name=name):
        return {"content": [{"type": "text", "text": f"{_name} of {args['order_id']}: 4"}]}
    return handler


depot = create_sdk_mcp_server("depot", tools=[depot_tool(*row) for row in ROWS])

_name=name binds the row's name into each handler, so all ten handlers do not close over the last row. With the shop's lookup and the desk's own built-ins, a session now has a dozen tools.

Holding definitions back

shop_desk.py changes in four places: __init__ reads the environment, in_context and search are new, and turn and run are replaced.

python

        # Claude Code defers by default; the desk waits to be asked, so a run stays short
        self.deferring = self.options.env.get("ENABLE_TOOL_SEARCH", "false").startswith(
            ("true", "auto"))
        self.loaded = set()

python

    def in_context(self):
        # what the model has definitions for: everything, or the searcher plus what it found
        if not self.deferring:
            return dict(self.tools)
        found = {name: self.tools[name] for name in self.loaded}
        return {"ToolSearch": ("built-in", "ToolSearch"), **found}

That is the whole idea. The model is handed in_context() rather than every tool the desk knows about, so with search on it starts with one definition, the searcher's.

python

    def search(self, query):
        words = set(query.lower().split())

        # the search matches the query against tool names, as the docs describe
        hits = [name for name in self.tools
                if words & set(name.lower().replace("mcp__", "").split("_"))]

        # five at most, and each one's definition stays in context from now on
        self.loaded.update(hits[:5])
        return "Loaded: " + ", ".join(hits[:5]) if hits else "No tool matched."

The search matches the query's words against tool names, keeps at most five, and remembers them. The real one matches names and descriptions, which is why a name like search_slack_messages surfaces for more requests than query_slack.

python

    async def turn(self, prompt):
        await self.start(prompt)
        results, turns = [], 0

        # the model is given the definitions in context, not every tool the desk has
        while "calls" in (step := self.model.reply(self.said, results, self.in_context())):
            turns += 1
            results = [await self.use(name, args) for name, args in step["calls"]]

        await self.say(step["text"])
        await self.finish("success", turns + 1, step["text"])

run answers the searcher itself, before it looks a tool up in the catalogue:

python

    async def run(self, name, args, call):
        # the searcher is a tool like any other, and it answers from the desk's catalogue
        if name == "ToolSearch":
            return self.search(args["query"]), False

ShopModel changes too: two branches are added to its reply, so that the model searches for a tool whose definition is not in context yet, and answers from a tool result rather than from a search result.

python

WANTS = {"refund": "refund_order", "stock": "stock_level"}

python

    def reply(self, said, results, tools):
        if self.script:
            return self.script.pop(0)

        # a search result is not an answer, so a searching turn does not stop here
        if results and not self.searched:
            return {"text": " ".join(results)}
        orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
        wanted = next((name for word, name in WANTS.items() if word in said[-1].lower()),
                      "lookup_order")
        tool = next((name for name in tools if name.endswith(wanted)), None)

python

        # the tool this message needs has no definition in context: search for it first
        if orders and tool is None and not self.searched and "ToolSearch" in tools:
            self.searched = True
            return {"calls": [("ToolSearch", {"query": wanted.replace("_", " ")})]}

        # the definition is in context now, so the call itself can be made
        if orders and tool:
            self.searched = False
            return {"calls": [(tool, {"order_id": order}) for order in orders]}
        return {"text": f"I cannot help with order {orders[0]} yet." if orders
                else "Which order is this about?"}

self.searched is the flag that keeps the two apart, and ShopModel.__init__ starts it at False. A search result arrives in results like any tool result, and without the flag the model would read it as the answer and stop.

python

    def __init__(self, script=None):
        self.script = list(script or [])
        self.searched = False

Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports these files. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep them in the same folder.

View the code here

shop_desk.py

import asyncio
import itertools
import json
import os
import re

from claude_agent_sdk import (AssistantMessage, ClaudeAgentOptions, ResultMessage,
                              SystemMessage, TextBlock, ToolResultBlock, ToolUseBlock,
                              Transport, UserMessage, query)

ORDER = re.compile(r"\b[A-Z]\d+\b")
SESSIONS = itertools.count(1)
WANTS = {"refund": "refund_order", "stock": "stock_level"}


def read_file(args):
    """What the desk serves for a Read call: the file's lines, numbered."""
    lines = open(args["file_path"]).read().splitlines()
    return "\n".join(f"{number:>5}  {line}" for number, line in enumerate(lines, 1))


BUILT_IN = {"Read": read_file}


class ShopModel:
    """Stands in for Claude: reads the conversation, picks the next step."""

    def __init__(self, script=None):
        self.script = list(script or [])
        self.searched = False

    def reply(self, said, results, tools):
        if self.script:
            return self.script.pop(0)

        # a search result is not an answer, so a searching turn does not stop here
        if results and not self.searched:
            return {"text": " ".join(results)}
        orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
        wanted = next((name for word, name in WANTS.items() if word in said[-1].lower()),
                      "lookup_order")
        tool = next((name for name in tools if name.endswith(wanted)), None)

        # the tool this message needs has no definition in context: search for it first
        if orders and tool is None and not self.searched and "ToolSearch" in tools:
            self.searched = True
            return {"calls": [("ToolSearch", {"query": wanted.replace("_", " ")})]}

        # the definition is in context now, so the call itself can be made
        if orders and tool:
            self.searched = False
            return {"calls": [(tool, {"order_id": order}) for order in orders]}
        return {"text": f"I cannot help with order {orders[0]} yet." if orders
                else "Which order is this about?"}


class ShopDesk(Transport):
    """Stands in for the Claude Code binary: it runs the loop over JSON lines."""

    def __init__(self, options=None, model=None):
        self.options = options or ClaudeAgentOptions()
        self.model = model or ShopModel()
        self.session_id = f"session-{next(SESSIONS)}"
        self.inbox, self.outbox = asyncio.Queue(), asyncio.Queue()
        self.prompts, self.said = asyncio.Queue(), []
        self.ids, self.setup = itertools.count(1), {}
        self.ready, self.tools = False, {}
        self.waiting = {}
        self.external, self.servers = {}, []

        # Claude Code defers by default; the desk waits to be asked, so a run stays short
        self.deferring = self.options.env.get("ENABLE_TOOL_SEARCH", "false").startswith(
            ("true", "auto"))
        self.loaded = set()

    async def connect(self):
        self.ready = True

    def is_ready(self):
        return self.ready

    async def write(self, data):
        # one JSON line at a time, which is what the SDK writes to a real process
        await self.inbox.put(json.loads(data))

    async def end_input(self):
        # for a real process this would close its standard input
        await self.inbox.put(None)

    async def close(self):
        self.ready = False

    async def send(self, message):
        await self.outbox.put(message)

    async def read_messages(self):
        # the listener and the worker run alongside the SDK's own read loop
        asyncio.create_task(self.listen())
        self.worker = asyncio.create_task(self.work())

        while (message := await self.outbox.get()) is not None:
            if isinstance(message, Exception):
                raise message
            yield message

    async def listen(self):
        while (message := await self.inbox.get()) is not None:
            # a control request is the SDK asking a question; initialize is first
            if message["type"] == "control_request":
                if message["request"]["subtype"] == "initialize":
                    self.setup = message["request"]
                await self.respond(message["request_id"])

            # a control response answers a question the desk itself asked
            elif message["type"] == "control_response":
                answer = message["response"]
                self.waiting.pop(answer["request_id"]).set_result(answer)
            elif message["type"] == "user":
                await self.prompts.put(message["message"]["content"])

        # the input has ended: finish the work in hand, then close the stream
        await self.prompts.put(None)
        await self.worker
        await self.send(None)

    async def respond(self, request_id):
        # the stand-in says yes to every request the SDK makes of it
        await self.send({"type": "control_response", "response": {
            "subtype": "success", "request_id": request_id, "response": {}}})

    async def ask(self, request):
        # a future per request, so the answer can be awaited where it was asked
        request_id = f"desk_{next(self.ids)}"
        self.waiting[request_id] = asyncio.get_running_loop().create_future()
        await self.send({"type": "control_request", "request_id": request_id,
                         "request": request})
        return await self.waiting[request_id]

    async def work(self):
        while (prompt := await self.prompts.get()) is not None:
            await self.turn(prompt)

    async def start(self, prompt):
        self.said.append(prompt)
        if not self.tools:
            self.tools = await self.find_tools()

        # the init message reports what each configured server is doing
        await self.send({"type": "system", "subtype": "init", "model": "shop-desk",
                         "session_id": self.session_id, "tools": list(self.tools),
                         "mcp_servers": self.servers})

    async def turn(self, prompt):
        await self.start(prompt)
        results, turns = [], 0

        # the model is given the definitions in context, not every tool the desk has
        while "calls" in (step := self.model.reply(self.said, results, self.in_context())):
            turns += 1
            results = [await self.use(name, args) for name, args in step["calls"]]

        await self.say(step["text"])
        await self.finish("success", turns + 1, step["text"])

    def offered(self):
        # the tools option is the availability layer: None means every built-in the
        # desk has, a list names the ones to keep, and [] leaves Claude none
        asked = self.options.tools
        if asked is None or isinstance(asked, dict):
            return list(BUILT_IN)
        return [name for name in asked if name in BUILT_IN]

    async def mcp(self, server, method, params):
        message = {"jsonrpc": "2.0", "id": next(self.ids), "method": method,
                   "params": params}
        answer = await self.ask({"subtype": "mcp_message", "server_name": server,
                                 "message": message})
        return answer["response"]["mcp_response"]["result"]

    async def connect_stdio(self, name, config):
        # Claude Code starts the command and talks MCP over its standard input
        process = await asyncio.create_subprocess_exec(
            config["command"], *config.get("args", []),
            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
            env={**os.environ, **config.get("env", {})})
        self.external[name] = process

        hello = await self.stdio(name, "initialize", {"protocolVersion": "2025-06-18"})
        if hello is None:
            return {"name": name, "status": "failed",
                    "error": "the server gave no answer to initialize"}
        return {"name": name, "status": "connected", "serverInfo": hello["serverInfo"]}

    async def stdio(self, name, method, params):
        process = self.external[name]
        request = {"jsonrpc": "2.0", "id": next(self.ids), "method": method,
                   "params": params}
        process.stdin.write((json.dumps(request) + "\n").encode())
        await process.stdin.drain()

        # a server that died writes nothing and its pipe reaches end of file
        line = await process.stdout.readline()
        return json.loads(line)["result"] if line else None

    async def list_from(self, name, config):
        # an in-process server is handed over by the SDK; a stdio one is a command to run
        if not isinstance(config, dict) or config.get("type") == "sdk":
            await self.mcp(name, "initialize", {"protocolVersion": "2025-06-18",
                           "capabilities": {}, "clientInfo": {"name": "desk", "version": "1"}})
            listed = await self.mcp(name, "tools/list", {})
            return {"name": name, "status": "connected", "tools": listed["tools"]}

        status = await self.connect_stdio(name, config)
        if status["status"] == "connected":
            status["tools"] = (await self.stdio(name, "tools/list", {}))["tools"]
        return status

    async def find_tools(self):
        tools = {name: ("built-in", name) for name in self.offered()}
        self.servers = []

        # one status per configured server, whichever kind of server it is
        for name, config in self.options.mcp_servers.items():
            status = await self.list_from(name, config)
            self.servers.append(status)
            for tool in status.pop("tools", []):
                tools[f"mcp__{name}__{tool['name']}"] = (name, tool["name"])

        return {name: where for name, where in tools.items()
                if name not in self.options.disallowed_tools}

    def in_context(self):
        # what the model has definitions for: everything, or the searcher plus what it found
        if not self.deferring:
            return dict(self.tools)
        found = {name: self.tools[name] for name in self.loaded}
        return {"ToolSearch": ("built-in", "ToolSearch"), **found}

    def search(self, query):
        words = set(query.lower().split())

        # the search matches the query against tool names, as the docs describe
        hits = [name for name in self.tools
                if words & set(name.lower().replace("mcp__", "").split("_"))]

        # five at most, and each one's definition stays in context from now on
        self.loaded.update(hits[:5])
        return "Loaded: " + ", ".join(hits[:5]) if hits else "No tool matched."

    async def use(self, name, args, parent=None):
        # one id ties the request and its result together, as the message pair does
        call = f"toolu_{next(self.ids)}"
        block = {"type": "tool_use", "id": call, "name": name, "input": args}
        await self.send(self.message("assistant", [block], parent))

        text, failed = await self.run(name, args, call)
        result = {"type": "tool_result", "tool_use_id": call, "content": text,
                  "is_error": failed}
        await self.send(self.message("user", [result], parent))
        return text

    async def run(self, name, args, call):
        # the searcher is a tool like any other, and it answers from the desk's catalogue
        if name == "ToolSearch":
            return self.search(args["query"]), False

        where, tool = self.tools[name]
        if where == "built-in":
            try:
                return BUILT_IN[tool](args), False
            except Exception as error:
                return f"{type(error).__name__}: {error}", True
        call_args = {"name": tool, "arguments": args}
        result = (await self.stdio(where, "tools/call", call_args) if where in self.external
                  else await self.mcp(where, "tools/call", call_args))
        return result["content"][0]["text"], result.get("isError", False)

    async def say(self, text):
        # with include_partial_messages on, the text goes out a word at a time
        if self.options.include_partial_messages:
            for word in text.split(" "):
                delta = {"type": "text_delta", "text": word + " "}
                await self.send({"type": "stream_event", "uuid": f"ev_{next(self.ids)}",
                                 "session_id": self.session_id, "event": {
                                     "type": "content_block_delta", "index": 0,
                                     "delta": delta}})

        # the whole message still follows, so code written earlier keeps working
        await self.send(self.message("assistant", [{"type": "text", "text": text}]))

    async def finish(self, subtype, turns, text):
        ok = subtype == "success"
        result = {"type": "result", "subtype": subtype, "is_error": not ok,
                  "duration_ms": 0, "duration_api_ms": 0, "num_turns": turns,
                  "session_id": self.session_id, "total_cost_usd": 0.0,
                  "usage": {"input_tokens": len(" ".join(self.said).split()),
                            "output_tokens": len(text.split())}}

        # cost is zero and usage counts words, because no model was called
        result.update({"result": text} if ok else {"errors": [text]})
        await self.send(result)

    def message(self, kind, content, parent=None):
        body = {"role": kind, "content": content}
        if kind == "assistant":
            body.update(model="shop-desk", id=f"msg_{next(self.ids)}")
        return {"type": kind, "message": body, "parent_tool_use_id": parent,
                "session_id": self.session_id}


def describe(message):
    # the two messages with no content blocks are read from their own fields
    if isinstance(message, SystemMessage):
        return f"system     {message.subtype}"
    if isinstance(message, ResultMessage):
        return f"result     {message.subtype}: {message.result or message.errors}"

    # a type this course does not use prints its name rather than failing
    if not isinstance(message, (AssistantMessage, UserMessage)):
        return type(message).__name__
    return describe_block(message.content[0])


def describe_block(block):
    if isinstance(block, TextBlock):
        return f"assistant  {block.text}"
    if isinstance(block, ToolUseBlock):
        return f"assistant  calls {block.name} {block.input}"

    # a tool result sits on the user's side, and its error flag is worth seeing
    if isinstance(block, ToolResultBlock):
        flag = " (error)" if block.is_error else ""
        return f"user       result{flag}: {block.content}"
    return type(block).__name__


async def watch(prompt, options, model=None):
    """Run one question through a fresh desk and print every message that comes back."""
    async for message in query(prompt=prompt, options=options,
                               transport=ShopDesk(options, model)):
        print(describe(message))

shop_tools.py

from claude_agent_sdk import create_sdk_mcp_server, tool

ORDERS = {"A17": "Order A17 shipped on 3 March."}


# the description is what Claude reads to decide this is the right tool
@tool("lookup_order", "Look up an order's shipping status by its id, such as A17.",
      {"order_id": str})
async def lookup_order(args):
    order = args["order_id"]

    # an unknown id is answered in words, not raised, so Claude can pass it on
    status = ORDERS.get(order, f"No order {order} was found.")
    return {"content": [{"type": "text", "text": status}]}


shop = create_sdk_mcp_server("shop", tools=[lookup_order])

The same question with search off and on

Both servers go into one session, which gives it a dozen tools:

python

from claude_agent_sdk import ClaudeAgentOptions
from depot_tools import depot
from shop_tools import shop
from shop_desk import watch

servers = {"shop": shop, "depot": depot}

Example

off = ClaudeAgentOptions(mcp_servers=servers)
await watch("How much stock of A17 is there?", off)

print("--- with tool search ---")
on = ClaudeAgentOptions(mcp_servers=servers, env={"ENABLE_TOOL_SEARCH": "true"})
await watch("How much stock of A17 is there?", on)

Output

system     init
assistant  calls mcp__depot__stock_level {'order_id': 'A17'}
user       result: stock_level of A17: 4
assistant  stock_level of A17: 4
result     success: stock_level of A17: 4
--- with tool search ---
system     init
assistant  calls ToolSearch {'query': 'stock level'}
user       result: Loaded: mcp__depot__stock_level
assistant  calls mcp__depot__stock_level {'order_id': 'A17'}
user       result: stock_level of A17: 4
assistant  stock_level of A17: 4
result     success: stock_level of A17: 4

What the extra round trip bought

With search off the model had every definition and called mcp__depot__stock_level at once: one round trip, twelve definitions in context.
With search on it called ToolSearch first, was given the one matching tool, and then made the same call: two round trips, two definitions in context.
The trade is the point. One extra round trip per search, against a smaller context on every turn. Below about ten tools, loading everything is usually quicker.

Counting what the model was given

Keeping the transport in a variable lets you read the two numbers after the run:

Example

from claude_agent_sdk import query
from shop_desk import ShopDesk

# the transport is built here rather than inside watch, so it can be read after
on = ClaudeAgentOptions(mcp_servers=servers, env={"ENABLE_TOOL_SEARCH": "true"})
desk = ShopDesk(on)
async for message in query(prompt="How much stock of A17 is there?", options=on,
                           transport=desk):
    pass

print("tools the session has:", len(desk.tools))
print("definitions in context:", sorted(desk.in_context()))

Output

tools the session has: 12
definitions in context: ['ToolSearch', 'mcp__depot__stock_level']

Twelve tools, two definitions. That is the narrowing tool search is for, and the loaded definition stays available for later turns until the messages where it was discovered are compacted away.

Turning it on and off for real

In Claude Code tool search is on by default. It is configured through the env option, which the Python SDK merges over the inherited environment, not through a flag:

`ENABLE_TOOL_SEARCH`	What happens
unset	Tool search is on, with fallbacks to loading upfront on older models, on a non-first-party `ANTHROPIC_BASE_URL`, and on an Azure-hosted Foundry deployment
`true`	Stays on through a non-first-party base URL, which fails on proxies that do not forward `tool_reference` blocks
`auto`	Counts the tokens of the definitions it could defer and switches on when they reach 10% of the context window
`auto:5`	The same with your own percentage; a lower number switches on sooner
`false`	Off. Every definition is loaded on every turn

The stand-in reverses that default: it defers only when you set true or auto, so the earlier lessons' runs stay as they were. The names and the meanings are the real ones.

Helping the search find the right tool

Name the tool for what it finds, not for the system it talks to. search_slack_messages matches more requests than query_slack.
Put the keywords in the description. "Search Slack messages by keyword, channel or date range" matches queries that "Query Slack" misses.
Say what categories exist in the system prompt, with the claude_code preset and append, so the agent knows what is worth searching for.
Pre-approve with a wildcard, mcp__enterprise-tools__*, so a tool the search loads is not then stopped by a permission prompt.

Loading everything against searching

	Definitions upfront	Tool search
Context per turn	Every definition, every turn	A summary, plus what was loaded
Round trips	None extra	One per search
Suits	Under about ten tools	Dozens to thousands
Ceiling	The context window	10,000 tools in the catalogue, five loaded per search

Watch out. Tool search is not available everywhere, and no ENABLE_TOOL_SEARCH value overrides that: a model on the SDK's unsupported list, an Azure-hosted Microsoft Foundry deployment, and a session with the HIPAA configuration all load definitions upfront instead. Setting CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS also keeps it off.

Previous: External MCP servers
Next: Permission modes
Reference: the tool search page and MCP tool search

Try it yourself

Set env={"ENABLE_TOOL_SEARCH": "true"} and ask about an order rather than stock. Which tools does the search load for lookup order?
Change search to keep two hits rather than five and read what the model is given.
Give one depot tool a name with no words in common with any question, then ask for it.

PreviousExternal MCP servers

Little by little, you're building something great.

Tool search for large tool sets

A depot with ten more tools

Holding definitions back

The same question with search off and on

What the extra round trip bought

Counting what the model was given

Turning it on and off for real

Helping the search find the right tool

Loading everything against searching

Related