ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat: Search transforms for tool discovery (#3154)

Browse source 
* feat: Add search transforms for tool discovery

RegexSearchTransform and BM25SearchTransform collapse large tool
catalogs into a search interface so LLMs discover tools on demand
instead of receiving the full listing.

* chore: Update SDK documentation

* fix: call_tool recursion guard, atomic BM25 rebuild, hash includes descriptions

* Extract CatalogTransform base class for catalog-aware transforms

Transforms that replace list_tools() with synthetic components (like
search) need to read the real catalog at call time without triggering
their own replacement logic. CatalogTransform handles the re-entrant
bypass via per-instance ContextVar, exposing transform_tools() as the
subclass hook and get_tool_catalog() for catalog access.

* Add search transform examples for regex and BM25

* Add README for search transform examples

* Polish search example clients with rich output

* Remove hardcoded tool counts from search example subtitles

* Clarify that review bot feedback should be evaluated on its merits

* Expand search transform docs with proper hierarchy

---------

Co-authored-by: marvin-context-protocol[bot] <225465937+marvin-context-protocol[bot]@users.noreply.github.com>
This commit is contained in:
Jeremiah Lowin 2026-02-26 22:42:38 -05:00 • committed by GitHub
commit c96c0400f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 1798 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

2
CLAUDE.md
View file

@ -68,7 +68,7 @@ When modifying MCP functionality, changes typically need to be applied across al
- Keep commit messages brief - ideally just headlines, not detailed messages
- Focus on what changed, not how or why
- Always read issue comments for follow-up information (treat maintainers as authoritative)
- **Treat proposed solutions in issues skeptically.** The ideal issue contains a concise problem description and an MRE — nothing more. Proposed solutions are only worth considering if they clearly reflect genuine, non-obvious investigation of the codebase. If a solution reads like speculation, or like it was generated by an LLM without deep framework knowledge, ignore it and diagnose from the repro. Most reporters — human or AI — do not have sufficient understanding of FastMCP internals to correctly diagnose anything beyond a trivial bug. We can ask the same questions of an LLM when implementing; we don't need the reporter to do it for us, and a wrong diagnosis is worse than none.
- **Treat proposed solutions in issues skeptically.** This applies to solutions proposed by *users* in issue reports — not to feedback from configured review bots (CodeRabbit, chatgpt-codex-connector, etc.), which should be evaluated on their merits. The ideal issue contains a concise problem description and an MRE — nothing more. Proposed solutions are only worth considering if they clearly reflect genuine, non-obvious investigation of the codebase. If a solution reads like speculation, or like it was generated by an LLM without deep framework knowledge, ignore it and diagnose from the repro. Most reporters — human or AI — do not have sufficient understanding of FastMCP internals to correctly diagnose anything beyond a trivial bug. We can ask the same questions of an LLM when implementing; we don't need the reporter to do it for us, and a wrong diagnosis is worse than none.
### PR Messages - Required Structure

1
docs/docs.json
View file

@ -163,6 +163,7 @@
"servers/transforms/namespace",
"servers/transforms/tool-transformation",
"servers/visibility",
"servers/transforms/tool-search",
"servers/transforms/resources-as-tools",
"servers/transforms/prompts-as-tools"
]

177
docs/servers/transforms/tool-search.mdx Normal file
View file

@ -0,0 +1,177 @@
---
title: Tool Search
sidebarTitle: Tool Search
description: Replace large tool catalogs with on-demand search
icon: magnifying-glass
tag: NEW
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.0.0" />
When a server exposes hundreds or thousands of tools, sending the full catalog to an LLM wastes tokens and degrades tool selection accuracy. Search transforms solve this by replacing the tool listing with a search interface — the LLM discovers tools on demand instead of receiving everything upfront.
## How It Works
When you add a search transform, `list_tools()` returns just two synthetic tools instead of the full catalog:
- **`search_tools`** finds tools matching a query and returns their full definitions
- **`call_tool`** executes a discovered tool by name
The original tools are still callable. They're hidden from the listing but remain fully functional — the search transform controls *discovery*, not *access*.
Both synthetic tools search across tool names, descriptions, parameter names, and parameter descriptions. A search for `"email"` would match a tool named `send_email`, a tool with "email" in its description, or a tool with an `email_address` parameter.
Search results are returned in the same JSON format as `list_tools`, including the full input schema, so the LLM can construct valid calls immediately without a second round-trip.
## Search Strategies
FastMCP provides two search transforms. They share the same interface — two synthetic tools, same configuration options — but differ in how they match queries to tools.
### Regex Search
`RegexSearchTransform` matches tools against a regex pattern using case-insensitive `re.search`. It has zero overhead and no index to build, making it a good default when the LLM knows roughly what it's looking for.
```python
from fastmcp import FastMCP
from fastmcp.server.transforms.search import RegexSearchTransform
mcp = FastMCP("My Server")
@mcp.tool
def search_database(query: str, limit: int = 10) -> list[dict]:
"""Search the database for records matching the query."""
...
@mcp.tool
def delete_record(record_id: str) -> bool:
"""Delete a record from the database by its ID."""
...
@mcp.tool
def send_email(to: str, subject: str, body: str) -> bool:
"""Send an email to the given recipient."""
...
mcp.add_transform(RegexSearchTransform())
```
The LLM's `search_tools` call takes a `pattern` parameter — a regex string:
```python
# Exact substring match
result = await client.call_tool("search_tools", {"pattern": "database"})
# Returns: search_database, delete_record
# Regex pattern
result = await client.call_tool("search_tools", {"pattern": "send.*email|notify"})
# Returns: send_email
```
Results are returned in catalog order. If the pattern is invalid regex, the search returns an empty list rather than raising an error.
### BM25 Search
`BM25SearchTransform` ranks tools by relevance using the [BM25 Okapi](https://en.wikipedia.org/wiki/Okapi_BM25) algorithm. It's better for natural language queries because it scores each tool based on term frequency and document rarity, returning results ranked by relevance rather than filtering by match/no-match.
```python
from fastmcp import FastMCP
from fastmcp.server.transforms.search import BM25SearchTransform
mcp = FastMCP("My Server")
# ... define tools ...
mcp.add_transform(BM25SearchTransform())
```
The LLM's `search_tools` call takes a `query` parameter — natural language:
```python
result = await client.call_tool("search_tools", {
"query": "tools for deleting things from the database"
})
# Returns: delete_record ranked first, search_database second
```
BM25 builds an in-memory index from the searchable text of all tools. The index is created lazily on the first search and automatically rebuilt whenever the tool catalog changes — for example, when tools are added, removed, or have their descriptions updated. The staleness check is based on a hash of all searchable text, so description changes are detected even when tool names stay the same.
### Which to Choose
Use **regex** when your LLM is good at constructing targeted patterns and you want deterministic, predictable results. Regex is also simpler to debug — you can see exactly what pattern was sent.
Use **BM25** when your LLM tends to describe what it needs in natural language, or when your tool catalog has nuanced descriptions where relevance ranking adds value. BM25 handles partial matches and synonyms better because it scores on individual terms rather than requiring a single pattern to match.
## Configuration
Both search transforms accept the same configuration options.
### Limiting Results
By default, search returns at most 5 tools. Adjust `max_results` based on your catalog size and how much context you want the LLM to receive per search:
```python
mcp.add_transform(RegexSearchTransform(max_results=10))
mcp.add_transform(BM25SearchTransform(max_results=3))
```
With regex, results stop as soon as the limit is reached (first N matches in catalog order). With BM25, all tools are scored and the top N by relevance are returned.
### Pinning Tools
Some tools should always be visible regardless of search. Use `always_visible` to pin them in the listing alongside the synthetic tools:
```python
mcp.add_transform(RegexSearchTransform(
always_visible=["help", "status"],
))
# list_tools returns: help, status, search_tools, call_tool
```
Pinned tools appear directly in `list_tools` so the LLM can call them without searching. They're excluded from search results to avoid duplication.
### Custom Tool Names
The default names `search_tools` and `call_tool` can be changed to avoid conflicts with real tools:
```python
mcp.add_transform(RegexSearchTransform(
search_tool_name="find_tools",
call_tool_name="run_tool",
))
```
## The `call_tool` Proxy
The `call_tool` proxy forwards calls to the real tool. When a client calls `call_tool(name="search_database", arguments={...})`, the proxy resolves `search_database` through the server's normal tool pipeline — including transforms and middleware — and executes it.
The proxy rejects attempts to call the synthetic tools themselves. `call_tool(name="call_tool")` raises an error rather than recursing.
<Note>
Tools discovered through search can also be called directly via `client.call_tool("search_database", {...})` without going through the proxy. The proxy exists for LLMs that only know about the tools returned by `list_tools` and need a way to invoke discovered tools through a tool they can see.
</Note>
## Auth and Visibility
Search results respect the full authorization pipeline. Tools filtered by middleware, visibility transforms, or component-level auth checks won't appear in search results.
The search tool queries `list_tools()` through the complete pipeline at search time, so the same filtering that controls what a client sees in the listing also controls what they can discover through search.
```python
from fastmcp.server.transforms import Visibility
from fastmcp.server.transforms.search import RegexSearchTransform
mcp = FastMCP("My Server")
# ... define tools ...
# Disable admin tools globally
mcp.add_transform(Visibility(False, tags={"admin"}))
# Add search — admin tools won't appear in results
mcp.add_transform(RegexSearchTransform())
```
Session-level visibility changes (via `ctx.disable_components()`) are also reflected immediately in search results.

1
docs/servers/transforms/transforms.mdx
View file

@ -29,6 +29,7 @@ FastMCP provides several transforms for common use cases:
- **[Namespace](/servers/transforms/namespace)** - Prefix component names to prevent conflicts when composing servers
- **[Tool Transformation](/servers/transforms/tool-transformation)** - Rename tools, modify descriptions, reshape arguments
- **[Enabled](/servers/visibility)** - Control which components are visible at runtime
- **[Tool Search](/servers/transforms/tool-search)** - Replace large tool catalogs with on-demand search
- **[Resources as Tools](/servers/transforms/resources-as-tools)** - Expose resources to tool-only clients
- **[Prompts as Tools](/servers/transforms/prompts-as-tools)** - Expose prompts to tool-only clients

66
examples/search/README.md Normal file
View file

@ -0,0 +1,66 @@
# Search Transforms
When a server exposes many tools, listing them all at once can overwhelm an LLM's context window. Search transforms collapse the full tool catalog behind a search interface — clients see only `search_tools` and `call_tool`, and discover the real tools on demand.
## Two search strategies
**Regex** (`RegexSearchTransform`) — clients search with regex patterns like `add|multiply` or `text.*`. Fast and precise when you know what you're looking for.
**BM25** (`BM25SearchTransform`) — clients search with natural language like `"work with numbers"`. Results are ranked by relevance using BM25 scoring. The index rebuilds automatically when tools change.
Both strategies respect the full auth pipeline: middleware, visibility transforms, and component-level auth checks all apply to search results.
## Run
```bash
# Regex
uv run python server_regex.py # in one terminal
uv run python client_regex.py # in another
# BM25
uv run python server_bm25.py
uv run python client_bm25.py
```
## Example Output (Regex)
```
=== Available Tools ===
- search_tools: Search for tools matching a regex pattern.
- call_tool: Call a tool by name with the given arguments.
=== Search: math tools (pattern: 'add|multiply|fibonacci') ===
- add: Add two numbers together.
- multiply: Multiply two numbers.
- fibonacci: Generate the first n Fibonacci numbers.
=== Search: text tools (pattern: 'text|string|word') ===
- reverse_string: Reverse a string.
- word_count: Count the number of words in a text.
- to_uppercase: Convert text to uppercase.
=== Calling 'add' via call_tool ===
Result: 42
```
## Example Output (BM25)
```
=== Available Tools ===
- list_files: List files in a directory.
- search_tools: Search for tools using natural language.
- call_tool: Call a tool by name with the given arguments.
=== Search: 'work with numbers' ===
- multiply: Multiply two numbers.
- add: Add two numbers together.
- fibonacci: Generate the first n Fibonacci numbers.
=== Search: 'file operations' ===
- read_file: Read the contents of a file.
=== Calling 'word_count' via call_tool ===
Result: 6
```
Note how `list_files` appears in the BM25 example's tool listing — it's pinned via `always_visible=["list_files"]`, keeping it visible alongside the synthetic search tools.

93
examples/search/client_bm25.py Normal file
View file

@ -0,0 +1,93 @@
"""Example: Client using BM25 search to discover and call tools.
BM25 search accepts natural language queries instead of regex patterns.
This client shows how relevance ranking surfaces the best matches.
Run with:
uv run python examples/search/client_bm25.py
"""
import asyncio
import json
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from fastmcp.client import Client
console = Console()
def _get_text(result) -> str:
"""Extract text content from a CallToolResult."""
return result.content[0].text
def _tool_table(tools: list[dict], *, ranked: bool = False) -> Table:
table = Table(show_header=True, show_edge=False, pad_edge=False, expand=True)
if ranked:
table.add_column("#", style="dim", width=3, justify="right")
table.add_column("Tool", style="cyan", no_wrap=True)
table.add_column("Description", style="dim")
for i, tool in enumerate(tools, 1):
row = [tool["name"], tool.get("description", "")]
if ranked:
row.insert(0, str(i))
table.add_row(*row)
return table
async def main():
async with Client("examples/search/server_bm25.py") as client:
console.print()
console.rule("[bold]BM25 Search Transform[/bold]")
console.print()
# list_files is pinned via always_visible
tools = await client.list_tools()
visible = [{"name": t.name, "description": t.description} for t in tools]
console.print(
Panel(
_tool_table(visible),
title="[bold]list_tools()[/bold]",
subtitle="[dim]list_files pinned via always_visible[/dim]",
border_style="blue",
)
)
console.print()
# Natural language searches — BM25 ranks by relevance
queries = ["work with numbers", "manipulate text strings", "file operations"]
for query in queries:
result = await client.call_tool("search_tools", {"query": query})
found = json.loads(_get_text(result))
console.print(
Panel(
_tool_table(found, ranked=True),
title=f'[bold]search_tools[/bold][dim](query="{query}")[/dim]',
subtitle=f"[dim]{len(found)} result{'s' if len(found) != 1 else ''}[/dim]",
border_style="green",
)
)
console.print()
# Call a discovered tool
result = await client.call_tool(
"call_tool",
{
"name": "word_count",
"arguments": {"text": "BM25 search makes tool discovery easy"},
},
)
call_label = Text.assemble(
("call_tool", "bold"),
('(word_count, text="BM25 search makes tool discovery easy")', "dim"),
)
console.print(call_label, "→", f"[bold green]{_get_text(result)}[/bold green]")
console.print()
if __name__ == "__main__":
asyncio.run(main())

105
examples/search/client_regex.py Normal file
View file

@ -0,0 +1,105 @@
"""Example: Client using regex search to discover and call tools.
Demonstrates the workflow: list tools (sees only search_tools + call_tool),
search for tools matching a regex pattern, then call a discovered tool.
Run with:
uv run python examples/search/client_regex.py
"""
import asyncio
import json
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from fastmcp.client import Client
console = Console()
def _get_text(result) -> str:
"""Extract text content from a CallToolResult."""
return result.content[0].text
def _tool_table(tools: list[dict]) -> Table:
table = Table(show_header=True, show_edge=False, pad_edge=False, expand=True)
table.add_column("Tool", style="cyan", no_wrap=True)
table.add_column("Description", style="dim")
for tool in tools:
table.add_row(tool["name"], tool.get("description", ""))
return table
async def main():
async with Client("examples/search/server_regex.py") as client:
console.print()
console.rule("[bold]Regex Search Transform[/bold]")
console.print()
# Show what the client actually sees
tools = await client.list_tools()
visible = [{"name": t.name, "description": t.description} for t in tools]
console.print(
Panel(
_tool_table(visible),
title="[bold]list_tools()[/bold]",
subtitle="[dim]real tools are discoverable via search[/dim]",
border_style="blue",
)
)
console.print()
# Search for math tools
result = await client.call_tool(
"search_tools", {"pattern": "add|multiply|fibonacci"}
)
found = json.loads(_get_text(result))
console.print(
Panel(
_tool_table(found),
title='[bold]search_tools[/bold][dim](pattern="add|multiply|fibonacci")[/dim]',
border_style="green",
)
)
console.print()
# Search for text tools
result = await client.call_tool("search_tools", {"pattern": "text|string|word"})
found = json.loads(_get_text(result))
console.print(
Panel(
_tool_table(found),
title='[bold]search_tools[/bold][dim](pattern="text|string|word")[/dim]',
border_style="green",
)
)
console.print()
# Call discovered tools via the proxy
result = await client.call_tool(
"call_tool", {"name": "add", "arguments": {"a": 17, "b": 25}}
)
call_label = Text.assemble(
("call_tool", "bold"),
("(add, a=17, b=25)", "dim"),
)
console.print(call_label, "→", f"[bold green]{_get_text(result)}[/bold green]")
result = await client.call_tool(
"call_tool",
{"name": "reverse_string", "arguments": {"text": "hello world"}},
)
call_label = Text.assemble(
("call_tool", "bold"),
('(reverse_string, text="hello world")', "dim"),
)
console.print(call_label, "→", f"[bold green]{_get_text(result)}[/bold green]")
console.print()
if __name__ == "__main__":
asyncio.run(main())

85
examples/search/server_bm25.py Normal file
View file

@ -0,0 +1,85 @@
"""Example: Search transforms with BM25 relevance ranking.
BM25SearchTransform uses term-frequency/inverse-document-frequency scoring
to rank tools by relevance to a natural language query. Unlike regex search
(which requires the user to construct a pattern), BM25 handles queries like
"work with text" or "do math" and returns the most relevant matches.
The index is built lazily and rebuilt automatically when the tool catalog
changes (e.g. tools added or removed between requests).
Run with:
uv run python examples/search/server_bm25.py
"""
from fastmcp import FastMCP
from fastmcp.server.transforms.search import BM25SearchTransform
mcp = FastMCP("BM25 Search Demo")
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@mcp.tool
def multiply(x: float, y: float) -> float:
"""Multiply two numbers."""
return x * y
@mcp.tool
def fibonacci(n: int) -> list[int]:
"""Generate the first n Fibonacci numbers."""
if n <= 0:
return []
seq = [0, 1]
while len(seq) < n:
seq.append(seq[-1] + seq[-2])
return seq[:n]
@mcp.tool
def reverse_string(text: str) -> str:
"""Reverse a string."""
return text[::-1]
@mcp.tool
def word_count(text: str) -> int:
"""Count the number of words in a text."""
return len(text.split())
@mcp.tool
def to_uppercase(text: str) -> str:
"""Convert text to uppercase."""
return text.upper()
@mcp.tool
def list_files(directory: str) -> list[str]:
"""List files in a directory."""
import os
return os.listdir(directory)
@mcp.tool
def read_file(path: str) -> str:
"""Read the contents of a file."""
with open(path) as f:
return f.read()
# BM25 search with a higher result limit for this larger catalog.
# The `always_visible` option keeps specific tools in list_tools output
# alongside the search/call tools — useful for tools the LLM should
# always know about.
mcp.add_transform(BM25SearchTransform(max_results=5, always_visible=["list_files"]))
if __name__ == "__main__":
mcp.run()

74
examples/search/server_regex.py Normal file
View file

@ -0,0 +1,74 @@
"""Example: Search transforms with regex pattern matching.
When a server has many tools, listing them all at once can overwhelm an LLM's
context window. Search transforms collapse the full tool catalog behind a
search interface — clients see only `search_tools` and `call_tool`, and
discover the real tools on demand.
This example registers a handful of tools and applies RegexSearchTransform.
Clients use `search_tools` with a regex pattern to find relevant tools, then
`call_tool` to execute them by name.
Run with:
uv run python examples/search/server_regex.py
"""
from fastmcp import FastMCP
from fastmcp.server.transforms.search import RegexSearchTransform
mcp = FastMCP("Regex Search Demo")
# Register a variety of tools across different domains.
# With the search transform active, none of these appear in list_tools —
# they're only discoverable via search.
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@mcp.tool
def multiply(x: float, y: float) -> float:
"""Multiply two numbers."""
return x * y
@mcp.tool
def fibonacci(n: int) -> list[int]:
"""Generate the first n Fibonacci numbers."""
if n <= 0:
return []
seq = [0, 1]
while len(seq) < n:
seq.append(seq[-1] + seq[-2])
return seq[:n]
@mcp.tool
def reverse_string(text: str) -> str:
"""Reverse a string."""
return text[::-1]
@mcp.tool
def word_count(text: str) -> int:
"""Count the number of words in a text."""
return len(text.split())
@mcp.tool
def to_uppercase(text: str) -> str:
"""Convert text to uppercase."""
return text.upper()
# Apply the regex search transform.
# max_results limits how many tools a single search returns.
mcp.add_transform(RegexSearchTransform(max_results=3))
if __name__ == "__main__":
mcp.run()

4
src/fastmcp/server/transforms/__init__.py
View file

@ -228,10 +228,6 @@ from fastmcp.server.transforms.tool_transform import ToolTransform # noqa: E402
from fastmcp.server.transforms.version_filter import VersionFilter # noqa: E402
__all__ = [
"GetPromptNext",
"GetResourceNext",
"GetResourceTemplateNext",
"GetToolNext",
"Namespace",
"PromptsAsTools",
"ResourcesAsTools",

239
src/fastmcp/server/transforms/catalog.py Normal file
View file

@ -0,0 +1,239 @@
"""Base class for transforms that need to read the real component catalog.
Some transforms replace ``list_tools()`` output with synthetic components
(e.g. a search interface) while still needing access to the *real*
(auth-filtered) catalog at call time. ``CatalogTransform`` provides the
bypass machinery so subclasses can call ``get_tool_catalog()`` without
triggering their own replacement logic.
Re-entrancy problem
-------------------
When a synthetic tool handler calls ``get_tool_catalog()``, that calls
``ctx.fastmcp.list_tools()`` which re-enters the transform pipeline —
including *this* transform's ``list_tools()``. If the subclass overrides
``list_tools()`` directly, the re-entrant call would hit the subclass's
replacement logic again (returning synthetic tools instead of the real
catalog). A ``super()`` call can't prevent this because Python can't
short-circuit a method after ``super()`` returns.
Solution: ``CatalogTransform`` owns ``list_tools()`` and uses a
per-instance ``ContextVar`` to detect re-entrant calls. During bypass,
it passes through to the base ``Transform.list_tools()`` (a no-op).
Otherwise, it delegates to ``transform_tools()`` — the subclass hook
where replacement logic lives. Same pattern for resources, prompts,
and resource templates.
This is *not* the same as the ``Provider._list_tools()`` convention
(which produces raw components with no arguments). ``transform_tools()``
receives the current catalog and returns a transformed version. The
distinct name avoids confusion between the two patterns.
Usage::
class MyTransform(CatalogTransform):
async def transform_tools(self, tools):
return [self._make_search_tool()]
def _make_search_tool(self):
async def search(ctx: Context = None):
real_tools = await self.get_tool_catalog(ctx)
...
return Tool.from_function(fn=search, name="search")
"""
from __future__ import annotations
import itertools
from collections.abc import Sequence
from contextvars import ContextVar
from typing import TYPE_CHECKING
from fastmcp.server.transforms import Transform
if TYPE_CHECKING:
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.context import Context
from fastmcp.tools.tool import Tool
_instance_counter = itertools.count()
class CatalogTransform(Transform):
"""Transform that needs access to the real component catalog.
Subclasses override ``transform_tools()`` / ``transform_resources()``
/ ``transform_prompts()`` / ``transform_resource_templates()``
instead of the ``list_*()`` methods. The base class owns
``list_*()`` and handles re-entrant bypass automatically — subclasses
never see re-entrant calls from ``get_*_catalog()``.
The ``get_*_catalog()`` methods fetch the real (auth-filtered) catalog
by temporarily setting a bypass flag so that this transform's
``list_*()`` passes through without calling the subclass hook.
"""
def __init__(self) -> None:
self._instance_id: int = next(_instance_counter)
self._bypass: ContextVar[bool] = ContextVar(
f"_catalog_bypass_{self._instance_id}", default=False
)
# ------------------------------------------------------------------
# list_* (bypass-aware — subclasses override transform_* instead)
# ------------------------------------------------------------------
async def list_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
if self._bypass.get():
return await super().list_tools(tools)
return await self.transform_tools(tools)
async def list_resources(self, resources: Sequence[Resource]) -> Sequence[Resource]:
if self._bypass.get():
return await super().list_resources(resources)
return await self.transform_resources(resources)
async def list_resource_templates(
self, templates: Sequence[ResourceTemplate]
) -> Sequence[ResourceTemplate]:
if self._bypass.get():
return await super().list_resource_templates(templates)
return await self.transform_resource_templates(templates)
async def list_prompts(self, prompts: Sequence[Prompt]) -> Sequence[Prompt]:
if self._bypass.get():
return await super().list_prompts(prompts)
return await self.transform_prompts(prompts)
# ------------------------------------------------------------------
# Subclass hooks (override these, not list_*)
# ------------------------------------------------------------------
async def transform_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
"""Transform the tool catalog.
Override this method to replace, filter, or augment the tool listing.
The default implementation passes through unchanged.
Do NOT override ``list_tools()`` directly — the base class uses it
to handle re-entrant bypass when ``get_tool_catalog()`` reads the
real catalog.
"""
return tools
async def transform_resources(
self, resources: Sequence[Resource]
) -> Sequence[Resource]:
"""Transform the resource catalog.
Override this method to replace, filter, or augment the resource listing.
The default implementation passes through unchanged.
Do NOT override ``list_resources()`` directly — the base class uses it
to handle re-entrant bypass when ``get_resource_catalog()`` reads the
real catalog.
"""
return resources
async def transform_resource_templates(
self, templates: Sequence[ResourceTemplate]
) -> Sequence[ResourceTemplate]:
"""Transform the resource template catalog.
Override this method to replace, filter, or augment the template listing.
The default implementation passes through unchanged.
Do NOT override ``list_resource_templates()`` directly — the base class
uses it to handle re-entrant bypass when
``get_resource_template_catalog()`` reads the real catalog.
"""
return templates
async def transform_prompts(self, prompts: Sequence[Prompt]) -> Sequence[Prompt]:
"""Transform the prompt catalog.
Override this method to replace, filter, or augment the prompt listing.
The default implementation passes through unchanged.
Do NOT override ``list_prompts()`` directly — the base class uses it
to handle re-entrant bypass when ``get_prompt_catalog()`` reads the
real catalog.
"""
return prompts
# ------------------------------------------------------------------
# Catalog accessors
# ------------------------------------------------------------------
async def get_tool_catalog(
self, ctx: Context, *, run_middleware: bool = True
) -> Sequence[Tool]:
"""Fetch the real tool catalog, bypassing this transform.
Args:
ctx: The current request context.
run_middleware: Whether to run middleware on the inner call.
Defaults to True because this is typically called from a
tool handler where list_tools middleware has not yet run.
"""
token = self._bypass.set(True)
try:
return await ctx.fastmcp.list_tools(run_middleware=run_middleware)
finally:
self._bypass.reset(token)
async def get_resource_catalog(
self, ctx: Context, *, run_middleware: bool = True
) -> Sequence[Resource]:
"""Fetch the real resource catalog, bypassing this transform.
Args:
ctx: The current request context.
run_middleware: Whether to run middleware on the inner call.
Defaults to True because this is typically called from a
tool handler where list_resources middleware has not yet run.
"""
token = self._bypass.set(True)
try:
return await ctx.fastmcp.list_resources(run_middleware=run_middleware)
finally:
self._bypass.reset(token)
async def get_prompt_catalog(
self, ctx: Context, *, run_middleware: bool = True
) -> Sequence[Prompt]:
"""Fetch the real prompt catalog, bypassing this transform.
Args:
ctx: The current request context.
run_middleware: Whether to run middleware on the inner call.
Defaults to True because this is typically called from a
tool handler where list_prompts middleware has not yet run.
"""
token = self._bypass.set(True)
try:
return await ctx.fastmcp.list_prompts(run_middleware=run_middleware)
finally:
self._bypass.reset(token)
async def get_resource_template_catalog(
self, ctx: Context, *, run_middleware: bool = True
) -> Sequence[ResourceTemplate]:
"""Fetch the real resource template catalog, bypassing this transform.
Args:
ctx: The current request context.
run_middleware: Whether to run middleware on the inner call.
Defaults to True because this is typically called from a
tool handler where list_resource_templates middleware has
not yet run.
"""
token = self._bypass.set(True)
try:
return await ctx.fastmcp.list_resource_templates(
run_middleware=run_middleware
)
finally:
self._bypass.reset(token)

23
src/fastmcp/server/transforms/search/__init__.py Normal file
View file

@ -0,0 +1,23 @@
"""Search transforms for tool discovery.
Search transforms collapse a large tool catalog into a search interface,
letting LLMs discover tools on demand instead of seeing the full list.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.transforms.search import RegexSearchTransform
mcp = FastMCP("Server")
mcp.add_transform(RegexSearchTransform())
# list_tools now returns only search_tools + call_tool
```
"""
from fastmcp.server.transforms.search.bm25 import BM25SearchTransform
from fastmcp.server.transforms.search.regex import RegexSearchTransform
__all__ = [
"BM25SearchTransform",
"RegexSearchTransform",
]

171
src/fastmcp/server/transforms/search/base.py Normal file
View file

@ -0,0 +1,171 @@
"""Base class for search transforms.
Search transforms replace ``list_tools()`` output with a small set of
synthetic tools — a search tool and a call-tool proxy — so LLMs can
discover tools on demand instead of receiving the full catalog.
All concrete search transforms (``RegexSearchTransform``,
``BM25SearchTransform``, etc.) inherit from ``BaseSearchTransform`` and
implement ``_make_search_tool()`` and ``_search()`` to provide their
specific search strategy.
Example::
from fastmcp import FastMCP
from fastmcp.server.transforms.search import RegexSearchTransform
mcp = FastMCP("Server")
@mcp.tool
def add(a: int, b: int) -> int: ...
@mcp.tool
def multiply(x: float, y: float) -> float: ...
# Clients now see only ``search_tools`` and ``call_tool``.
# The original tools are discoverable via search.
mcp.add_transform(RegexSearchTransform())
"""
from abc import abstractmethod
from collections.abc import Sequence
from typing import Annotated, Any
from fastmcp.server.context import Context
from fastmcp.server.transforms import GetToolNext
from fastmcp.server.transforms.catalog import CatalogTransform
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities.versions import VersionSpec
def _extract_searchable_text(tool: Tool) -> str:
"""Combine tool name, description, and parameter info into searchable text."""
parts = [tool.name]
if tool.description:
parts.append(tool.description)
schema = tool.parameters
if schema:
properties = schema.get("properties", {})
for param_name, param_info in properties.items():
parts.append(param_name)
if isinstance(param_info, dict):
desc = param_info.get("description", "")
if desc:
parts.append(desc)
return " ".join(parts)
def _serialize_tools_for_output(tools: Sequence[Tool]) -> list[dict[str, Any]]:
"""Serialize tools to the same dict format as ``list_tools`` output."""
return [
tool.to_mcp_tool().model_dump(mode="json", exclude_none=True) for tool in tools
]
class BaseSearchTransform(CatalogTransform):
"""Replace the tool listing with a search interface.
When this transform is active, ``list_tools()`` returns only:
* Any tools listed in ``always_visible`` (pinned).
* A **search tool** that finds tools matching a query.
* A **call_tool** proxy that executes tools discovered via search.
Hidden tools remain callable — ``get_tool()`` delegates unknown
names downstream, so direct calls and the call-tool proxy both work.
Search results respect the full auth pipeline: middleware, visibility
transforms, and component-level auth checks all apply.
Args:
max_results: Maximum number of tools returned per search.
always_visible: Tool names that stay in the ``list_tools``
output alongside the synthetic search/call tools.
search_tool_name: Name of the generated search tool.
call_tool_name: Name of the generated call-tool proxy.
"""
def __init__(
self,
*,
max_results: int = 5,
always_visible: list[str] | None = None,
search_tool_name: str = "search_tools",
call_tool_name: str = "call_tool",
) -> None:
super().__init__()
self._max_results = max_results
self._always_visible = set(always_visible or [])
self._search_tool_name = search_tool_name
self._call_tool_name = call_tool_name
# ------------------------------------------------------------------
# Transform interface
# ------------------------------------------------------------------
async def transform_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
"""Replace the catalog with pinned + synthetic search/call tools."""
pinned = [t for t in tools if t.name in self._always_visible]
return [*pinned, self._make_search_tool(), self._make_call_tool()]
async def get_tool(
self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None
) -> Tool | None:
"""Intercept synthetic tool names; delegate everything else."""
if name == self._search_tool_name:
return self._make_search_tool()
if name == self._call_tool_name:
return self._make_call_tool()
return await call_next(name, version=version)
# ------------------------------------------------------------------
# Synthetic tools
# ------------------------------------------------------------------
@abstractmethod
def _make_search_tool(self) -> Tool:
"""Create the search tool. Subclasses define the parameter schema."""
...
def _make_call_tool(self) -> Tool:
"""Create the call_tool proxy that executes discovered tools."""
transform = self
async def call_tool(
name: Annotated[str, "The name of the tool to call"],
arguments: Annotated[
dict[str, Any] | None, "Arguments to pass to the tool"
] = None,
ctx: Context = None, # type: ignore[assignment]
) -> ToolResult:
"""Call a tool by name with the given arguments.
Use this to execute tools discovered via search_tools.
"""
if name in {transform._call_tool_name, transform._search_tool_name}:
raise ValueError(
f"'{name}' is a synthetic search tool and cannot be called via the call_tool proxy"
)
return await ctx.fastmcp.call_tool(name, arguments)
return Tool.from_function(fn=call_tool, name=self._call_tool_name)
# ------------------------------------------------------------------
# Catalog access
# ------------------------------------------------------------------
async def _get_visible_tools(self, ctx: Context) -> Sequence[Tool]:
"""Get the auth-filtered tool catalog, excluding pinned tools."""
tools = await self.get_tool_catalog(ctx)
return [t for t in tools if t.name not in self._always_visible]
# ------------------------------------------------------------------
# Abstract search
# ------------------------------------------------------------------
@abstractmethod
async def _search(self, tools: Sequence[Tool], query: str) -> Sequence[Tool]:
"""Search the given tools and return matches."""
...

142
src/fastmcp/server/transforms/search/bm25.py Normal file
View file

@ -0,0 +1,142 @@
"""BM25-based search transform."""
import hashlib
import math
import re
from collections.abc import Sequence
from typing import Annotated, Any
from fastmcp.server.context import Context
from fastmcp.server.transforms.search.base import (
BaseSearchTransform,
_extract_searchable_text,
_serialize_tools_for_output,
)
from fastmcp.tools.tool import Tool
def _tokenize(text: str) -> list[str]:
"""Lowercase, split on non-alphanumeric, filter short tokens."""
return [t for t in re.split(r"[^a-z0-9]+", text.lower()) if len(t) > 1]
class _BM25Index:
"""Self-contained BM25 Okapi index."""
def __init__(self, k1: float = 1.5, b: float = 0.75) -> None:
self.k1 = k1
self.b = b
self._doc_tokens: list[list[str]] = []
self._doc_lengths: list[int] = []
self._avg_dl: float = 0.0
self._df: dict[str, int] = {}
self._tf: list[dict[str, int]] = []
self._n: int = 0
def build(self, documents: list[str]) -> None:
self._doc_tokens = [_tokenize(doc) for doc in documents]
self._doc_lengths = [len(tokens) for tokens in self._doc_tokens]
self._n = len(documents)
self._avg_dl = sum(self._doc_lengths) / self._n if self._n else 0.0
self._df = {}
self._tf = []
for tokens in self._doc_tokens:
tf: dict[str, int] = {}
seen: set[str] = set()
for token in tokens:
tf[token] = tf.get(token, 0) + 1
if token not in seen:
self._df[token] = self._df.get(token, 0) + 1
seen.add(token)
self._tf.append(tf)
def query(self, text: str, top_k: int) -> list[int]:
"""Return indices of top_k documents sorted by BM25 score."""
query_tokens = _tokenize(text)
if not query_tokens or not self._n:
return []
scores: list[float] = [0.0] * self._n
for token in query_tokens:
if token not in self._df:
continue
idf = math.log(
(self._n - self._df[token] + 0.5) / (self._df[token] + 0.5) + 1.0
)
for i in range(self._n):
tf = self._tf[i].get(token, 0)
if tf == 0:
continue
dl = self._doc_lengths[i]
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * dl / self._avg_dl)
scores[i] += idf * numerator / denominator
ranked = sorted(range(self._n), key=lambda i: scores[i], reverse=True)
return [i for i in ranked[:top_k] if scores[i] > 0]
def _catalog_hash(tools: Sequence[Tool]) -> str:
"""SHA256 hash of sorted tool searchable text for staleness detection."""
key = "|".join(sorted(_extract_searchable_text(t) for t in tools))
return hashlib.sha256(key.encode()).hexdigest()
class BM25SearchTransform(BaseSearchTransform):
"""Search transform using BM25 Okapi relevance ranking.
Maintains an in-memory index that is lazily rebuilt when the tool
catalog changes (detected via a hash of tool names).
"""
def __init__(
self,
*,
max_results: int = 5,
always_visible: list[str] | None = None,
search_tool_name: str = "search_tools",
call_tool_name: str = "call_tool",
) -> None:
super().__init__(
max_results=max_results,
always_visible=always_visible,
search_tool_name=search_tool_name,
call_tool_name=call_tool_name,
)
self._index = _BM25Index()
self._indexed_tools: Sequence[Tool] = ()
self._last_hash: str = ""
def _make_search_tool(self) -> Tool:
transform = self
async def search_tools(
query: Annotated[str, "Natural language query to search for tools"],
ctx: Context = None, # type: ignore[assignment]
) -> list[dict[str, Any]]:
"""Search for tools using natural language.
Returns matching tool definitions ranked by relevance,
in the same format as list_tools.
"""
hidden = await transform._get_visible_tools(ctx)
results = await transform._search(hidden, query)
return _serialize_tools_for_output(results)
return Tool.from_function(fn=search_tools, name=self._search_tool_name)
async def _search(self, tools: Sequence[Tool], query: str) -> Sequence[Tool]:
current_hash = _catalog_hash(tools)
if current_hash != self._last_hash:
documents = [_extract_searchable_text(t) for t in tools]
new_index = _BM25Index(self._index.k1, self._index.b)
new_index.build(documents)
self._index, self._indexed_tools, self._last_hash = (
new_index,
tools,
current_hash,
)
indices = self._index.query(query, self._max_results)
return [self._indexed_tools[i] for i in indices]

56
src/fastmcp/server/transforms/search/regex.py Normal file
View file

@ -0,0 +1,56 @@
"""Regex-based search transform."""
import re
from collections.abc import Sequence
from typing import Annotated, Any
from fastmcp.server.context import Context
from fastmcp.server.transforms.search.base import (
BaseSearchTransform,
_extract_searchable_text,
_serialize_tools_for_output,
)
from fastmcp.tools.tool import Tool
class RegexSearchTransform(BaseSearchTransform):
"""Search transform using regex pattern matching.
Tools are matched against their name, description, and parameter
information using ``re.search`` with ``re.IGNORECASE``.
"""
def _make_search_tool(self) -> Tool:
transform = self
async def search_tools(
pattern: Annotated[
str,
"Regex pattern to match against tool names, descriptions, and parameters",
],
ctx: Context = None, # type: ignore[assignment]
) -> list[dict[str, Any]]:
"""Search for tools matching a regex pattern.
Returns matching tool definitions in the same format as list_tools.
"""
hidden = await transform._get_visible_tools(ctx)
results = await transform._search(hidden, pattern)
return _serialize_tools_for_output(results)
return Tool.from_function(fn=search_tools, name=self._search_tool_name)
async def _search(self, tools: Sequence[Tool], query: str) -> Sequence[Tool]:
try:
compiled = re.compile(query, re.IGNORECASE)
except re.error:
return []
matches: list[Tool] = []
for tool in tools:
text = _extract_searchable_text(tool)
if compiled.search(text):
matches.append(tool)
if len(matches) >= self._max_results:
break
return matches

82
tests/server/transforms/test_catalog.py Normal file
View file

@ -0,0 +1,82 @@
"""Tests for CatalogTransform base class."""
from __future__ import annotations
from collections.abc import Sequence
from mcp.types import TextContent
from fastmcp import FastMCP
from fastmcp.server.context import Context
from fastmcp.server.transforms import GetToolNext
from fastmcp.server.transforms.catalog import CatalogTransform
from fastmcp.tools.tool import Tool
from fastmcp.utilities.versions import VersionSpec
class ReplacingTransform(CatalogTransform):
"""Minimal subclass that replaces tools with a synthetic tool.
Uses ``get_tool_catalog()`` to read the real catalog inside the
synthetic tool's handler, verifying that the bypass mechanism works.
"""
async def transform_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
return [self._make_synthetic_tool()]
async def get_tool(
self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None
) -> Tool | None:
if name == "count_tools":
return self._make_synthetic_tool()
return await call_next(name, version=version)
def _make_synthetic_tool(self) -> Tool:
transform = self
async def count_tools(ctx: Context = None) -> int: # type: ignore[assignment]
"""Return the number of real tools in the catalog."""
catalog = await transform.get_tool_catalog(ctx)
return len(catalog)
return Tool.from_function(fn=count_tools, name="count_tools")
class TestCatalogTransformBypass:
async def test_list_tools_replaced_by_subclass(self):
mcp = FastMCP("test")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
@mcp.tool
def multiply(x: float, y: float) -> float:
return x * y
mcp.add_transform(ReplacingTransform())
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert names == {"count_tools"}
async def test_get_tool_catalog_returns_real_tools(self):
mcp = FastMCP("test")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
@mcp.tool
def multiply(x: float, y: float) -> float:
return x * y
mcp.add_transform(ReplacingTransform())
result = await mcp.call_tool("count_tools", {})
assert any("2" in c.text for c in result.content if isinstance(c, TextContent))
async def test_multiple_instances_have_independent_bypass(self):
"""Each CatalogTransform instance has its own bypass ContextVar."""
t1 = ReplacingTransform()
t2 = ReplacingTransform()
assert t1._instance_id != t2._instance_id
assert t1._bypass is not t2._bypass

482
tests/server/transforms/test_search.py Normal file
View file

@ -0,0 +1,482 @@
"""Tests for search transforms."""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
from unittest.mock import MagicMock
import mcp.types as mcp_types
import pytest
from mcp.types import TextContent
from fastmcp import Client, FastMCP
from fastmcp.server.context import Context
from fastmcp.server.middleware.middleware import CallNext, Middleware, MiddlewareContext
from fastmcp.server.transforms import Visibility
from fastmcp.server.transforms.search.bm25 import (
BM25SearchTransform,
_BM25Index,
_catalog_hash,
)
from fastmcp.server.transforms.search.regex import RegexSearchTransform
from fastmcp.tools.tool import Tool, ToolResult
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _parse_tool_result(result: ToolResult) -> list[dict[str, Any]]:
"""Extract tool list from a ToolResult's structured content."""
assert result.structured_content is not None
return result.structured_content["result"]
def _make_server_with_tools() -> FastMCP:
mcp = FastMCP("test")
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@mcp.tool
def multiply(x: float, y: float) -> float:
"""Multiply two numbers."""
return x * y
@mcp.tool
def search_database(query: str, limit: int = 10) -> str:
"""Search the database for records matching the query."""
return f"results for {query}"
@mcp.tool
def delete_record(record_id: str) -> str:
"""Delete a record from the database by its ID."""
return f"deleted {record_id}"
@mcp.tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return "sent"
return mcp
# ---------------------------------------------------------------------------
# Shared behavior tests (parameterized across both transforms)
# ---------------------------------------------------------------------------
class TestBaseTransformBehavior:
"""Tests for behavior shared by all search transforms."""
async def test_list_tools_hides_tools_regex(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert names == {"search_tools", "call_tool"}
async def test_list_tools_hides_tools_bm25(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert names == {"search_tools", "call_tool"}
async def test_always_visible_pins_tools(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform(always_visible=["add"]))
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert "add" in names
assert "search_tools" in names
assert "call_tool" in names
assert "multiply" not in names
async def test_get_tool_returns_synthetic(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
search = await mcp.get_tool("search_tools")
assert search is not None
assert search.name == "search_tools"
call = await mcp.get_tool("call_tool")
assert call is not None
assert call.name == "call_tool"
async def test_get_tool_passes_through_hidden(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
tool = await mcp.get_tool("add")
assert tool is not None
assert tool.name == "add"
async def test_call_tool_proxy_executes(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
# Need to call list_tools to populate catalog
await mcp.list_tools()
result = await mcp.call_tool(
"call_tool", {"name": "add", "arguments": {"a": 2, "b": 3}}
)
assert any("5" in c.text for c in result.content if isinstance(c, TextContent))
async def test_custom_tool_names(self):
mcp = _make_server_with_tools()
mcp.add_transform(
RegexSearchTransform(
search_tool_name="find_tools",
call_tool_name="run_tool",
)
)
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert names == {"find_tools", "run_tool"}
assert await mcp.get_tool("find_tools") is not None
assert await mcp.get_tool("run_tool") is not None
async def test_search_respects_visibility_filtering(self):
"""Tools disabled via Visibility transform should not appear in search."""
mcp = _make_server_with_tools()
mcp.add_transform(Visibility(False, names={"delete_record"}))
mcp.add_transform(RegexSearchTransform())
tools = await mcp.list_tools()
names = {t.name for t in tools}
assert "delete_record" not in names
result = await mcp.call_tool("search_tools", {"pattern": "delete"})
found = _parse_tool_result(result)
assert not any(t["name"] == "delete_record" for t in found)
async def test_search_respects_auth_middleware(self):
"""Tools filtered by auth middleware should not appear in search."""
class BlockAdminTools(Middleware):
async def on_list_tools(
self,
context: MiddlewareContext[mcp_types.ListToolsRequest],
call_next: CallNext[mcp_types.ListToolsRequest, Sequence[Tool]],
) -> Sequence[Tool]:
tools = await call_next(context)
return [t for t in tools if t.name != "delete_record"]
mcp = _make_server_with_tools()
mcp.add_middleware(BlockAdminTools())
mcp.add_transform(RegexSearchTransform())
async with Client(mcp) as client:
tools = await client.list_tools()
names = {t.name for t in tools}
assert "delete_record" not in names
assert "search_tools" in names
result = await client.call_tool("search_tools", {"pattern": "delete"})
found = _parse_tool_result(result)
assert not any(t["name"] == "delete_record" for t in found)
async def test_search_respects_session_visibility(self):
"""Tools disabled via session visibility should not appear in search."""
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
@mcp.tool
async def disable_delete(ctx: Context) -> str:
"""Helper tool to disable delete_record for this session."""
await ctx.disable_components(names={"delete_record"})
return "disabled"
async with Client(mcp) as client:
# Before disabling, search should find delete_record
result = await client.call_tool("search_tools", {"pattern": "delete"})
found = _parse_tool_result(result)
assert any(t["name"] == "delete_record" for t in found)
# Disable via session visibility
await client.call_tool("disable_delete", {})
# After disabling, search should NOT find it
result = await client.call_tool("search_tools", {"pattern": "delete"})
found = _parse_tool_result(result)
assert not any(t["name"] == "delete_record" for t in found)
# ---------------------------------------------------------------------------
# Regex-specific tests
# ---------------------------------------------------------------------------
class TestRegexSearch:
async def test_search_by_name(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "add"})
tools = _parse_tool_result(result)
assert any(t["name"] == "add" for t in tools)
async def test_search_by_description(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "email"})
tools = _parse_tool_result(result)
assert any(t["name"] == "send_email" for t in tools)
async def test_search_by_param_name(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "record_id"})
tools = _parse_tool_result(result)
assert any(t["name"] == "delete_record" for t in tools)
async def test_search_by_param_description(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "recipient"})
tools = _parse_tool_result(result)
assert any(t["name"] == "send_email" for t in tools)
async def test_search_or_pattern(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform(max_results=10))
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "add|multiply"})
tools = _parse_tool_result(result)
names = {t["name"] for t in tools}
assert "add" in names
assert "multiply" in names
async def test_search_case_insensitive(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "ADD"})
tools = _parse_tool_result(result)
assert any(t["name"] == "add" for t in tools)
async def test_search_invalid_pattern(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "[invalid"})
tools = _parse_tool_result(result)
assert tools == []
async def test_search_max_results(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform(max_results=2))
await mcp.list_tools()
# Match everything
result = await mcp.call_tool("search_tools", {"pattern": ".*"})
tools = _parse_tool_result(result)
assert len(tools) == 2
async def test_search_no_matches(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "zzz_nonexistent"})
tools = _parse_tool_result(result)
assert tools == []
async def test_search_returns_full_schema(self):
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"pattern": "add"})
tools = _parse_tool_result(result)
add_tool = next(t for t in tools if t["name"] == "add")
assert "inputSchema" in add_tool
assert "properties" in add_tool["inputSchema"]
# ---------------------------------------------------------------------------
# BM25-specific tests
# ---------------------------------------------------------------------------
class TestBM25Search:
async def test_search_relevance(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": "database"})
tools = _parse_tool_result(result)
# Database tools should rank highest
assert len(tools) > 0
names = {t["name"] for t in tools}
assert "search_database" in names
async def test_search_database_tools(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
result = await mcp.call_tool(
"search_tools", {"query": "delete records from database"}
)
tools = _parse_tool_result(result)
assert len(tools) > 0
# delete_record should be highly relevant
assert tools[0]["name"] == "delete_record"
async def test_search_max_results(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform(max_results=2))
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": "number"})
tools = _parse_tool_result(result)
assert len(tools) <= 2
async def test_search_no_matches(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": "zzz_nonexistent_xyz"})
tools = _parse_tool_result(result)
assert tools == []
async def test_index_rebuilds_on_catalog_change(self):
"""When a new tool is added, the next list_tools + search sees it."""
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
# Search before adding tool
result = await mcp.call_tool("search_tools", {"query": "weather forecast"})
tools = _parse_tool_result(result)
assert not any(t["name"] == "get_weather" for t in tools)
# Add a new tool
@mcp.tool
def get_weather(city: str) -> str:
"""Get the weather forecast for a city."""
return f"sunny in {city}"
# Must call list_tools to refresh the catalog cache
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": "weather forecast"})
tools = _parse_tool_result(result)
assert any(t["name"] == "get_weather" for t in tools)
async def test_search_empty_query(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": ""})
tools = _parse_tool_result(result)
assert tools == []
async def test_search_returns_full_schema(self):
mcp = _make_server_with_tools()
mcp.add_transform(BM25SearchTransform())
await mcp.list_tools()
result = await mcp.call_tool("search_tools", {"query": "add numbers"})
tools = _parse_tool_result(result)
add_tool = next(t for t in tools if t["name"] == "add")
assert "inputSchema" in add_tool
assert "properties" in add_tool["inputSchema"]
# ---------------------------------------------------------------------------
# BM25 index unit tests
# ---------------------------------------------------------------------------
class TestBM25Index:
def test_basic_ranking(self):
index = _BM25Index()
index.build(
[
"search database query records",
"add two numbers together",
"send email recipient subject",
]
)
results = index.query("database records", 3)
assert results[0] == 0 # Database doc should rank first
def test_empty_corpus(self):
index = _BM25Index()
index.build([])
assert index.query("anything", 5) == []
def test_no_matching_tokens(self):
index = _BM25Index()
index.build(["alpha beta gamma"])
assert index.query("zzz", 5) == []
# ---------------------------------------------------------------------------
# call_tool self-reference guard
# ---------------------------------------------------------------------------
class TestCallToolGuard:
async def test_call_tool_proxy_rejects_itself(self):
"""Calling call_tool(name='call_tool') must not recurse infinitely."""
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
async with Client(mcp) as client:
with pytest.raises(Exception):
await client.call_tool(
"call_tool", {"name": "call_tool", "arguments": {}}
)
async def test_call_tool_proxy_rejects_search_tool(self):
"""Calling call_tool(name='search_tools') must be rejected."""
mcp = _make_server_with_tools()
mcp.add_transform(RegexSearchTransform())
async with Client(mcp) as client:
with pytest.raises(Exception):
await client.call_tool(
"call_tool",
{"name": "search_tools", "arguments": {"pattern": "add"}},
)
async def test_call_tool_proxy_rejects_custom_names(self):
"""Guard works when synthetic tools have custom names."""
mcp = _make_server_with_tools()
mcp.add_transform(
RegexSearchTransform(
search_tool_name="find_tools", call_tool_name="run_tool"
)
)
async with Client(mcp) as client:
with pytest.raises(Exception):
await client.call_tool(
"run_tool", {"name": "run_tool", "arguments": {}}
)
with pytest.raises(Exception):
await client.call_tool(
"run_tool", {"name": "find_tools", "arguments": {"pattern": "add"}}
)
# ---------------------------------------------------------------------------
# catalog hash staleness
# ---------------------------------------------------------------------------
class TestCatalogHash:
def test_hash_differs_for_same_name_different_description(self):
"""Hash must change when a tool's description changes, not just its name."""
tool_a = MagicMock()
tool_a.name = "search"
tool_a.description = "find records in the database"
tool_a.parameters = {}
tool_b = MagicMock()
tool_b.name = "search"
tool_b.description = "send an email to a recipient"
tool_b.parameters = {}
assert _catalog_hash([tool_a]) != _catalog_hash([tool_b])