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

Add ListTools, search limit, and catalog size annotation to CodeMode (#3359)

Browse source 
* Add tests for two-stage pattern, empty full-detail results, empty inputs

* Add ListTools, search limit, catalog size annotation; split tests

Co-authored-by: Claude <noreply@anthropic.com>

* Remove BM25 internal cap so Search.limit is the sole truncation point

* Pass default_limit to BM25 instead of arbitrary high cap

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin 2026-03-02 18:00:21 -05:00 • committed by GitHub
commit 404b820144
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 704 additions and 264 deletions
Download patch file Download diff file Expand all files Collapse all files

42
docs/servers/transforms/code-mode.mdx
View file

@ -30,7 +30,7 @@ You take a normal server with normally registered tools and add a `CodeMode` tra
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
mcp = FastMCP("Server", transforms=[CodeMode()])
@ -114,7 +114,7 @@ This three-stage flow works well for most servers — each step pulls in only th
## Discovery Tools
CodeMode ships with three built-in discovery tools: `Search`, `GetSchemas`, and `GetTags`. By default, only `Search` and `GetSchemas` are enabled. Each tool supports a `default_detail` parameter that sets the default verbosity level, and the LLM can override the detail level on any individual call.
CodeMode ships with four built-in discovery tools: `Search`, `GetSchemas`, `GetTags`, and `ListTools`. By default, only `Search` and `GetSchemas` are enabled. Each tool supports a `default_detail` parameter that sets the default verbosity level, and the LLM can override the detail level on any individual call.
### Detail Levels
@ -132,6 +132,14 @@ CodeMode ships with three built-in discovery tools: `Search`, `GetSchemas`, and
`Search` finds tools by natural-language query using BM25 ranking. At its default `"brief"` detail, results include just tool names and descriptions — enough to decide which tools are worth inspecting further. The LLM can request `"detailed"` to get parameter schemas inline, or `"full"` for the complete JSON.
Search results include an annotation like `"2 of 10 tools:"` when the result set is smaller than the full catalog, so the LLM knows there are more tools to discover with different queries.
You can cap result count with `default_limit`. The LLM can also override the limit per call. This is useful for large catalogs where you want to keep search results focused:
```python
Search(default_limit=5) # return at most 5 results per search
```
If your tools use [tags](/servers/tools#tags), Search also accepts a `tags` parameter so the LLM can narrow results to specific categories before searching.
### GetSchemas
@ -150,6 +158,20 @@ If your tools use [tags](/servers/tools#tags), Search also accepts a `tags` para
`GetTags` isn't included in the defaults — add it when browsing by category would help the LLM orient itself in a large catalog. The LLM can browse tags first, then pass specific tags into Search to narrow results.
### ListTools
`ListTools` dumps the entire catalog at whatever detail level the LLM requests. It supports the same three detail levels as `Search` and `GetSchemas`, defaulting to `"brief"`.
`ListTools` isn't included in the defaults — for large catalogs, search-based discovery is more token-efficient. But for smaller catalogs (under ~20 tools), letting the LLM see everything upfront can be faster than multiple search round-trips:
```python
from fastmcp.experimental.transforms.code_mode import CodeMode, ListTools, GetSchemas
code_mode = CodeMode(
discovery_tools=[ListTools(), GetSchemas()],
)
```
## Discovery Patterns
The right discovery configuration depends on your server — how many tools you have and how complex their parameters are. It may be tempting to minimize round-trips by collapsing everything into fewer steps, but for the complex servers that benefit most from CodeMode, our experience is that staged discovery leads to better results. Flooding the LLM with detailed schemas for tools it doesn't end up using can hurt more than the extra round-trip costs. Each pattern below is a complete, copyable configuration.
@ -160,7 +182,7 @@ The default. The LLM searches for candidates, inspects schemas for the ones it w
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
mcp = FastMCP("Server", transforms=[CodeMode()])
```
@ -169,7 +191,7 @@ If your tools use [tags](/servers/tools#tags), add `GetTags` so the LLM can brow
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import GetTags, Search, GetSchemas
code_mode = CodeMode(
@ -185,7 +207,7 @@ Search returns parameter schemas inline, so the LLM can go straight from search
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import Search, GetSchemas
code_mode = CodeMode(
@ -203,7 +225,7 @@ Skip discovery entirely and bake tool instructions into the execute tool's descr
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
code_mode = CodeMode(
discovery_tools=[],
@ -225,7 +247,7 @@ Discovery tools are composable — you can mix the built-ins with your own. Each
Here's a minimal example:
```python
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import GetToolCatalog, GetSchemas
from fastmcp.server.context import Context
from fastmcp.tools.tool import Tool
@ -266,7 +288,8 @@ mcp = FastMCP("Server", transforms=[code_mode])
The default `MontySandboxProvider` can enforce execution limits — timeouts, memory caps, recursion depth, and more. Without limits, LLM-generated scripts can run indefinitely.
```python
from fastmcp.experimental.transforms import CodeMode, MontySandboxProvider
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import MontySandboxProvider
sandbox = MontySandboxProvider(
limits={"max_duration_secs": 10, "max_memory": 50_000_000},
@ -293,7 +316,8 @@ You can replace the default sandbox with any object implementing the `SandboxPro
from collections.abc import Callable
from typing import Any
from fastmcp.experimental.transforms import CodeMode, SandboxProvider
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import SandboxProvider
class RemoteSandboxProvider:
async def run(

2
examples/code_mode/server.py
View file

@ -13,7 +13,7 @@ Run with:
"""
from fastmcp import FastMCP
from fastmcp.experimental.transforms import CodeMode
from fastmcp.experimental.transforms.code_mode import CodeMode
mcp = FastMCP("CodeMode Demo")

10
src/fastmcp/experimental/transforms/__init__.py
View file

@ -1,11 +1 @@
from .code_mode import (
CodeMode,
MontySandboxProvider,
SandboxProvider,
)
__all__ = [
"CodeMode",
"MontySandboxProvider",
"SandboxProvider",
]

64
src/fastmcp/experimental/transforms/code_mode.py
View file

@ -188,6 +188,8 @@ class Search:
``"brief"`` returns tool names and descriptions only.
``"detailed"`` returns compact markdown with parameter schemas.
``"full"`` returns complete JSON tool definitions.
default_limit: Maximum number of results to return.
The LLM can override this per call. ``None`` means no limit.
"""
def __init__(
@ -196,19 +198,22 @@ class Search:
search_fn: SearchFn | None = None,
name: str = "search",
default_detail: ToolDetailLevel | None = None,
default_limit: int | None = None,
) -> None:
if search_fn is None:
from fastmcp.server.transforms.search.bm25 import BM25SearchTransform
_bm25 = BM25SearchTransform()
_bm25 = BM25SearchTransform(max_results=default_limit or 50)
search_fn = _bm25._search
self._search_fn = search_fn
self._name = name
self._default_detail: ToolDetailLevel = default_detail or "brief"
self._default_limit = default_limit
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
search_fn = self._search_fn
default_detail = self._default_detail
default_limit = self._default_limit
async def search(
query: Annotated[str, "Search query to find available tools"],
@ -220,13 +225,19 @@ class Search:
ToolDetailLevel,
"'brief' for names and descriptions, 'detailed' for parameter schemas as markdown, 'full' for complete JSON schemas",
] = default_detail,
limit: Annotated[
int | None,
"Maximum number of results to return",
] = default_limit,
ctx: Context = None, # type: ignore[assignment]
) -> str:
"""Search for available tools by query.
Returns matching tools ranked by relevance.
"""
tools = await get_catalog(ctx)
catalog = await get_catalog(ctx)
catalog_size = len(catalog)
tools: Sequence[Tool] = catalog
if tags:
tag_set = set(tags)
has_untagged = "untagged" in tag_set
@ -237,7 +248,13 @@ class Search:
if (t.tags & real_tags) or (has_untagged and not t.tags)
]
results = await search_fn(tools, query)
return _render_tools(results, detail)
if limit is not None:
results = results[:limit]
rendered = _render_tools(results, detail)
if len(results) < catalog_size and detail != "full":
n = len(results)
rendered = f"{n} of {catalog_size} tools:\n\n{rendered}"
return rendered
return Tool.from_function(fn=search, name=self._name)
@ -370,6 +387,46 @@ class GetTags:
return Tool.from_function(fn=tags, name=self._name)
class ListTools:
"""Discovery tool factory that lists all tools in the catalog.
Args:
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level.
``"brief"`` returns tool names and one-line descriptions.
``"detailed"`` returns compact markdown with parameter schemas.
``"full"`` returns the complete JSON schema.
"""
def __init__(
self,
*,
name: str = "list_tools",
default_detail: ToolDetailLevel | None = None,
) -> None:
self._name = name
self._default_detail: ToolDetailLevel = default_detail or "brief"
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
default_detail = self._default_detail
async def list_tools(
detail: Annotated[
ToolDetailLevel,
"'brief' for names and descriptions, 'detailed' for parameter schemas as markdown, 'full' for complete JSON schemas",
] = default_detail,
ctx: Context = None, # type: ignore[assignment]
) -> str:
"""List all available tools.
Use to see the full catalog before searching or calling tools.
"""
catalog = await get_catalog(ctx)
return _render_tools(catalog, detail)
return Tool.from_function(fn=list_tools, name=self._name)
# ---------------------------------------------------------------------------
# CodeMode
# ---------------------------------------------------------------------------
@ -518,6 +575,7 @@ __all__ = [
"GetSchemas",
"GetTags",
"GetToolCatalog",
"ListTools",
"MontySandboxProvider",
"SandboxProvider",
"Search",

240
tests/experimental/transforms/test_code_mode.py
View file

@ -7,11 +7,11 @@ from mcp.types import ImageContent, TextContent
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.experimental.transforms import CodeMode, MontySandboxProvider
from fastmcp.experimental.transforms.code_mode import (
CodeMode,
GetSchemas,
GetTags,
GetToolCatalog,
MontySandboxProvider,
Search,
_ensure_async,
)
@ -470,242 +470,6 @@ def test_code_mode_rejects_duplicate_discovery_names() -> None:
cm._build_discovery_tools()
# ---------------------------------------------------------------------------
# Tags discovery tool
# ---------------------------------------------------------------------------
async def test_categories_brief_shows_tag_counts() -> None:
mcp = FastMCP("Tags Brief")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
return x + y
@mcp.tool(tags={"math"})
def multiply(x: int, y: int) -> int:
return x * y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "math (2 tools)" in text
assert "text (1 tool)" in text
async def test_categories_full_lists_tools_per_tag() -> None:
mcp = FastMCP("Tags Full")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "### math" in text
assert "- add: Add two numbers." in text
assert "### text" in text
assert "- greet: Say hello." in text
async def test_categories_includes_untagged() -> None:
mcp = FastMCP("Tags Untagged")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
return x + y
@mcp.tool
def ping() -> str:
return "pong"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "math" in text
assert "untagged (1 tool)" in text
async def test_categories_tool_in_multiple_tags() -> None:
mcp = FastMCP("Tags Multi-tag")
@mcp.tool(tags={"math", "core"})
def add(x: int, y: int) -> int:
return x + y
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "### core" in text
assert "### math" in text
# Tool appears under both tags
assert text.count("- add") == 2
async def test_categories_detail_override_per_call() -> None:
"""LLM can override default_detail on a per-call basis."""
mcp = FastMCP("Tags Override")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add numbers."""
return x + y
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()], # default_detail="brief"
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
# Override to full
result = await _run_tool(mcp, "tags", {"detail": "full"})
text = _unwrap_string_result(result)
assert "### math" in text
assert "- add: Add numbers." in text
# ---------------------------------------------------------------------------
# Search with tags filtering
# ---------------------------------------------------------------------------
async def test_search_with_tags_filter() -> None:
mcp = FastMCP("Search Tags")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello", "tags": ["math"]})
text = _unwrap_string_result(result)
assert "add" in text
assert "greet" not in text
async def test_search_with_tags_filter_no_matches() -> None:
mcp = FastMCP("Search Tags Empty")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add", "tags": ["nonexistent"]})
text = _unwrap_string_result(result)
assert "add" not in text or "No tools" in text
async def test_search_without_tags_returns_all() -> None:
"""Search without tags parameter searches the full catalog."""
mcp = FastMCP("Search No Tags")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello"})
text = _unwrap_string_result(result)
assert "add" in text
assert "greet" in text
async def test_search_with_untagged_filter() -> None:
"""Search with tags=["untagged"] matches tools that have no tags."""
mcp = FastMCP("Search Untagged")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool
def ping() -> str:
"""Ping."""
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "ping add", "tags": ["untagged"]})
text = _unwrap_string_result(result)
assert "ping" in text
assert "add" not in text
async def test_get_schema_full_partial_match_returns_valid_json() -> None:
"""get_schema with detail=full and missing tools returns valid JSON."""
mcp = FastMCP("Schema Full Partial")
@mcp.tool
def square(x: int) -> int:
"""Compute the square."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "get_schema", {"tools": ["square", "nonexistent"], "detail": "full"}
)
text = _unwrap_string_result(result)
parsed = json.loads(text)
assert isinstance(parsed, list)
assert parsed[0]["name"] == "square"
assert parsed[-1] == {"not_found": ["nonexistent"]}
# ---------------------------------------------------------------------------
# Visibility and auth
# ---------------------------------------------------------------------------

604
tests/experimental/transforms/test_code_mode_discovery.py Normal file
View file

@ -0,0 +1,604 @@
import json
from typing import Any
from mcp.types import TextContent
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import (
CodeMode,
GetTags,
ListTools,
Search,
_ensure_async,
)
from fastmcp.tools.tool import ToolResult
def _unwrap_result(result: ToolResult) -> Any:
"""Extract the logical return value from a ToolResult."""
if result.structured_content is not None:
return result.structured_content
text_blocks = [
content.text for content in result.content if isinstance(content, TextContent)
]
if not text_blocks:
return None
if len(text_blocks) == 1:
try:
return json.loads(text_blocks[0])
except json.JSONDecodeError:
return text_blocks[0]
values: list[Any] = []
for text in text_blocks:
try:
values.append(json.loads(text))
except json.JSONDecodeError:
values.append(text)
return values
def _unwrap_string_result(result: ToolResult) -> str:
"""Extract a string result from a ToolResult."""
data = _unwrap_result(result)
if isinstance(data, dict) and "result" in data:
return data["result"]
assert isinstance(data, str)
return data
class _UnsafeTestSandboxProvider:
"""UNSAFE: Uses exec() for testing only. Never use in production."""
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Any] | None = None,
) -> Any:
namespace: dict[str, Any] = {}
if inputs:
namespace.update(inputs)
if external_functions:
namespace.update(
{key: _ensure_async(value) for key, value in external_functions.items()}
)
wrapped = "async def __test_main__():\n"
for line in code.splitlines():
wrapped += f" {line}\n"
if not code.strip():
wrapped += " return None\n"
exec(wrapped, namespace, namespace)
return await namespace["__test_main__"]()
async def _run_tool(
server: FastMCP, name: str, arguments: dict[str, Any]
) -> ToolResult:
return await server.call_tool(name, arguments)
# ---------------------------------------------------------------------------
# Tags discovery tool
# ---------------------------------------------------------------------------
async def test_categories_brief_shows_tag_counts() -> None:
mcp = FastMCP("Tags Brief")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
return x + y
@mcp.tool(tags={"math"})
def multiply(x: int, y: int) -> int:
return x * y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "math (2 tools)" in text
assert "text (1 tool)" in text
async def test_categories_full_lists_tools_per_tag() -> None:
mcp = FastMCP("Tags Full")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "### math" in text
assert "- add: Add two numbers." in text
assert "### text" in text
assert "- greet: Say hello." in text
async def test_categories_includes_untagged() -> None:
mcp = FastMCP("Tags Untagged")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
return x + y
@mcp.tool
def ping() -> str:
return "pong"
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "math" in text
assert "untagged (1 tool)" in text
async def test_categories_tool_in_multiple_tags() -> None:
mcp = FastMCP("Tags Multi-tag")
@mcp.tool(tags={"math", "core"})
def add(x: int, y: int) -> int:
return x + y
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "### core" in text
assert "### math" in text
# Tool appears under both tags
assert text.count("- add") == 2
async def test_categories_detail_override_per_call() -> None:
"""LLM can override default_detail on a per-call basis."""
mcp = FastMCP("Tags Override")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add numbers."""
return x + y
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()], # default_detail="brief"
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
# Override to full
result = await _run_tool(mcp, "tags", {"detail": "full"})
text = _unwrap_string_result(result)
assert "### math" in text
assert "- add: Add numbers." in text
async def test_get_tags_empty_catalog() -> None:
"""GetTags with no tools returns 'No tools available.'."""
mcp = FastMCP("CodeMode Empty Tags")
mcp.disable(names={"ping"}, components={"tool"})
mcp.add_transform(
CodeMode(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "tags", {})
text = _unwrap_string_result(result)
assert "No tools available" in text
# ---------------------------------------------------------------------------
# Search with tags filtering
# ---------------------------------------------------------------------------
async def test_search_with_tags_filter() -> None:
mcp = FastMCP("Search Tags")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello", "tags": ["math"]})
text = _unwrap_string_result(result)
assert "add" in text
assert "greet" not in text
async def test_search_with_tags_filter_no_matches() -> None:
mcp = FastMCP("Search Tags Empty")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add", "tags": ["nonexistent"]})
text = _unwrap_string_result(result)
assert "add" not in text or "No tools" in text
async def test_search_without_tags_returns_all() -> None:
"""Search without tags parameter searches the full catalog."""
mcp = FastMCP("Search No Tags")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool(tags={"text"})
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello"})
text = _unwrap_string_result(result)
assert "add" in text
assert "greet" in text
async def test_search_with_untagged_filter() -> None:
"""Search with tags=["untagged"] matches tools that have no tags."""
mcp = FastMCP("Search Untagged")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool
def ping() -> str:
"""Ping."""
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "ping add", "tags": ["untagged"]})
text = _unwrap_string_result(result)
assert "ping" in text
assert "add" not in text
async def test_search_default_detail_detailed_skips_get_schema() -> None:
"""Two-stage pattern: Search(default_detail='detailed') returns schemas inline."""
mcp = FastMCP("CodeMode Two-Stage")
@mcp.tool
def square(x: int) -> int:
"""Compute the square."""
return x * x
mcp.add_transform(
CodeMode(
discovery_tools=[Search(default_detail="detailed")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "search", {"query": "square"})
text = _unwrap_string_result(result)
assert "### square" in text
assert "**Parameters**" in text
assert "`x` (integer, required)" in text
async def test_search_full_detail_empty_results_returns_json() -> None:
"""Search with detail=full and no matches returns valid JSON, not plain text."""
mcp = FastMCP("CodeMode Empty Full")
@mcp.tool(tags={"math"})
def add(x: int, y: int) -> int:
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp,
"search",
{"query": "nonexistent", "tags": ["nonexistent"], "detail": "full"},
)
text = _unwrap_string_result(result)
parsed = json.loads(text)
assert parsed == []
async def test_get_schema_empty_tools_list() -> None:
"""get_schema with an empty tools list returns no-match message."""
mcp = FastMCP("CodeMode Empty Schema")
@mcp.tool
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": []})
text = _unwrap_string_result(result)
assert "No tools matched" in text
async def test_get_schema_full_partial_match_returns_valid_json() -> None:
"""get_schema with detail=full and missing tools returns valid JSON."""
mcp = FastMCP("Schema Full Partial")
@mcp.tool
def square(x: int) -> int:
"""Compute the square."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "get_schema", {"tools": ["square", "nonexistent"], "detail": "full"}
)
text = _unwrap_string_result(result)
parsed = json.loads(text)
assert isinstance(parsed, list)
assert parsed[0]["name"] == "square"
assert parsed[-1] == {"not_found": ["nonexistent"]}
# ---------------------------------------------------------------------------
# Search catalog size annotation
# ---------------------------------------------------------------------------
async def test_search_shows_catalog_size_when_results_are_subset() -> None:
"""Search annotates results with 'N of M tools' when not all tools match."""
mcp = FastMCP("Search Annotation")
@mcp.tool
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool
def multiply(x: int, y: int) -> int:
"""Multiply two numbers."""
return x * y
@mcp.tool
def greet(name: str) -> str:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add numbers"})
text = _unwrap_string_result(result)
# Should show partial result count out of total catalog
assert "of 3 tools:" in text
async def test_search_omits_annotation_when_all_tools_returned() -> None:
"""Search does not annotate when results include every tool."""
mcp = FastMCP("Search All Match")
@mcp.tool
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add numbers"})
text = _unwrap_string_result(result)
assert "of" not in text or "tools:" not in text
# ---------------------------------------------------------------------------
# Search limit
# ---------------------------------------------------------------------------
async def test_search_limit_caps_results() -> None:
"""Search with limit returns at most that many results."""
mcp = FastMCP("Search Limit")
@mcp.tool
def add(x: int, y: int) -> int:
"""Add numbers."""
return x + y
@mcp.tool
def subtract(x: int, y: int) -> int:
"""Subtract numbers."""
return x - y
@mcp.tool
def multiply(x: int, y: int) -> int:
"""Multiply numbers."""
return x * y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "numbers", "limit": 1})
text = _unwrap_string_result(result)
assert "1 of 3 tools:" in text
# Only one tool line (starts with "- ")
tool_lines = [line for line in text.splitlines() if line.startswith("- ")]
assert len(tool_lines) == 1
async def test_search_default_limit_from_constructor() -> None:
"""Search(default_limit=N) caps results by default."""
mcp = FastMCP("Search Default Limit")
@mcp.tool
def a() -> str:
"""Tool A."""
return "a"
@mcp.tool
def b() -> str:
"""Tool B."""
return "b"
@mcp.tool
def c() -> str:
"""Tool C."""
return "c"
mcp.add_transform(
CodeMode(
discovery_tools=[Search(default_limit=2)],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "search", {"query": "tool"})
text = _unwrap_string_result(result)
assert "2 of 3 tools:" in text
# ---------------------------------------------------------------------------
# ListTools discovery tool
# ---------------------------------------------------------------------------
async def test_list_tools_brief() -> None:
"""ListTools at brief detail lists all tool names and descriptions."""
mcp = FastMCP("ListTools Brief")
@mcp.tool
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
@mcp.tool
def multiply(x: int, y: int) -> int:
"""Multiply two numbers."""
return x * y
mcp.add_transform(
CodeMode(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "list_tools", {})
text = _unwrap_string_result(result)
assert "add" in text
assert "multiply" in text
assert "Add two numbers" in text
# brief should not include parameter details
assert "**Parameters**" not in text
async def test_list_tools_detailed() -> None:
"""ListTools at detailed shows parameter schemas."""
mcp = FastMCP("ListTools Detailed")
@mcp.tool
def square(x: int) -> int:
"""Compute the square."""
return x * x
mcp.add_transform(
CodeMode(
discovery_tools=[ListTools(default_detail="detailed")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "list_tools", {})
text = _unwrap_string_result(result)
assert "### square" in text
assert "**Parameters**" in text
assert "`x` (integer, required)" in text
async def test_list_tools_full_returns_json() -> None:
"""ListTools at full returns valid JSON with schemas."""
mcp = FastMCP("ListTools Full")
@mcp.tool
def ping() -> str:
"""Ping."""
return "pong"
mcp.add_transform(
CodeMode(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "list_tools", {"detail": "full"})
text = _unwrap_string_result(result)
parsed = json.loads(text)
assert isinstance(parsed, list)
assert parsed[0]["name"] == "ping"
async def test_list_tools_empty_catalog() -> None:
"""ListTools with no tools returns no-match message."""
mcp = FastMCP("ListTools Empty")
mcp.add_transform(
CodeMode(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
)
result = await _run_tool(mcp, "list_tools", {})
text = _unwrap_string_result(result)
assert "No tools matched" in text

6
uv.lock generated
View file

@ -39,7 +39,7 @@ wheels = [
[[package]]
name = "anthropic"
version = "0.79.0"
version = "0.84.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@ -51,9 +51,9 @@ dependencies = [
{ name = "sniffio" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/15/b1/91aea3f8fd180d01d133d931a167a78a3737b3fd39ccef2ae8d6619c24fd/anthropic-0.79.0.tar.gz", hash = "sha256:8707aafb3b1176ed6c13e2b1c9fb3efddce90d17aee5d8b83a86c70dcdcca871", size = 509825, upload-time = "2026-02-07T18:06:18.388Z" }
sdist = { url = "https://files.pythonhosted.org/packages/04/ea/0869d6df9ef83dcf393aeefc12dd81677d091c6ffc86f783e51cf44062f2/anthropic-0.84.0.tar.gz", hash = "sha256:72f5f90e5aebe62dca316cb013629cfa24996b0f5a4593b8c3d712bc03c43c37", size = 539457, upload-time = "2026-02-25T05:22:38.54Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/95/b2/cc0b8e874a18d7da50b0fda8c99e4ac123f23bf47b471827c5f6f3e4a767/anthropic-0.79.0-py3-none-any.whl", hash = "sha256:04cbd473b6bbda4ca2e41dd670fe2f829a911530f01697d0a1e37321eb75f3cf", size = 405918, upload-time = "2026-02-07T18:06:20.246Z" },
{ url = "https://files.pythonhosted.org/packages/64/ca/218fa25002a332c0aa149ba18ffc0543175998b1f65de63f6d106689a345/anthropic-0.84.0-py3-none-any.whl", hash = "sha256:861c4c50f91ca45f942e091d83b60530ad6d4f98733bfe648065364da05d29e7", size = 455156, upload-time = "2026-02-25T05:22:40.468Z" },
]
[[package]]