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

Convert code-mode to the CodeMode plugin (#4002)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-04-22 09:46:29 -04:00 • committed by GitHub
commit 19fa2fc33e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1019 additions and 645 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -10,9 +10,9 @@ import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.1.0" />
<Warning>
CodeMode is experimental. The core interface is stable, but the specific discovery tools and their parameters may evolve as we learn more about what works best in practice.
</Warning>
<Note>
The core interface is stable, but the specific discovery tools and their parameters may evolve as we learn more about what works best in practice.
</Note>
Standard MCP tool usage has two scaling problems. First, every tool in the catalog is loaded into the LLM's context upfront — with hundreds of tools, that's tens of thousands of tokens spent before the LLM even reads the user's request. Second, every tool call is a round-trip: the LLM calls a tool, the result passes back through the context window, the LLM reasons about it, calls another tool, and so on. Intermediate results that only exist to feed the next step still burn tokens flowing through the model.
@ -26,13 +26,13 @@ The approach was introduced by Cloudflare in [Code Mode](https://blog.cloudflare
CodeMode requires the `code-mode` extra for sandbox support. Install it with `pip install "fastmcp[code-mode]"`.
</Tip>
You take a normal server with normally registered tools and add a `CodeMode` transform. The transform wraps your existing tools in the code mode machinery — your tool functions don't change at all:
You take a normal server with normally registered tools and attach the `CodeMode` plugin. The plugin wraps your existing tools in the code mode machinery — your tool functions don't change at all:
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import CodeMode
mcp = FastMCP("Server", transforms=[CodeMode()])
mcp = FastMCP("Server", plugins=[CodeMode()])
@mcp.tool
def add(x: int, y: int) -> int:
@ -165,7 +165,7 @@ If your tools use [tags](/servers/tools#tags), Search also accepts a `tags` para
`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
from fastmcp.server.plugins.code_mode import CodeMode, ListTools, GetSchemas
code_mode = CodeMode(
discovery_tools=[ListTools(), GetSchemas()],
@ -182,23 +182,23 @@ The default. The LLM searches for candidates, inspects schemas for the ones it w
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import CodeMode
mcp = FastMCP("Server", transforms=[CodeMode()])
mcp = FastMCP("Server", plugins=[CodeMode()])
```
If your tools use [tags](/servers/tools#tags), add `GetTags` so the LLM can browse by category before searching — giving it four stages of progressive disclosure:
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import GetTags, Search, GetSchemas
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import GetTags, Search, GetSchemas
code_mode = CodeMode(
discovery_tools=[GetTags(), Search(), GetSchemas()],
)
mcp = FastMCP("Server", transforms=[code_mode])
mcp = FastMCP("Server", plugins=[code_mode])
```
### Two-Stage
@ -207,14 +207,14 @@ Search returns parameter schemas inline, so the LLM can go straight from search
```python
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import Search, GetSchemas
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import Search, GetSchemas
code_mode = CodeMode(
discovery_tools=[Search(default_detail="detailed"), GetSchemas()],
)
mcp = FastMCP("Server", transforms=[code_mode])
mcp = FastMCP("Server", plugins=[code_mode])
```
`GetSchemas` is still available as a fallback — the LLM can call it with `detail="full"` if it encounters a tool with complex nested parameters where the compact markdown isn't enough.
@ -225,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.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import CodeMode
code_mode = CodeMode(
discovery_tools=[],
@ -237,7 +237,7 @@ code_mode = CodeMode(
),
)
mcp = FastMCP("Server", transforms=[code_mode])
mcp = FastMCP("Server", plugins=[code_mode])
```
## Custom Discovery Tools
@ -247,8 +247,8 @@ Discovery tools are composable — you can mix the built-ins with your own. Each
Here's a minimal example:
```python
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import GetToolCatalog, GetSchemas
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import GetToolCatalog, GetSchemas
from fastmcp.server.context import Context
from fastmcp.tools.tool import Tool
@ -268,7 +268,7 @@ The LLM sees the docstring of each discovery tool's inner function as its descri
Discovery tools and the execute tool can also have custom names:
```python
from fastmcp.experimental.transforms.code_mode import Search, GetSchemas
from fastmcp.server.plugins.code_mode import Search, GetSchemas
code_mode = CodeMode(
discovery_tools=[
@ -278,7 +278,7 @@ code_mode = CodeMode(
execute_tool_name="run_workflow",
)
mcp = FastMCP("Server", transforms=[code_mode])
mcp = FastMCP("Server", plugins=[code_mode])
```
## Sandbox Configuration
@ -288,14 +288,14 @@ 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.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import MontySandboxProvider
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import MontySandboxProvider
sandbox = MontySandboxProvider(
limits={"max_duration_secs": 10, "max_memory": 50_000_000},
)
mcp = FastMCP("Server", transforms=[CodeMode(sandbox_provider=sandbox)])
mcp = FastMCP("Server", plugins=[CodeMode(sandbox_provider=sandbox)])
```
All keys are optional — omit any to leave that dimension uncapped:
@ -316,8 +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.code_mode import CodeMode
from fastmcp.experimental.transforms.code_mode import SandboxProvider
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import SandboxProvider
class RemoteSandboxProvider:
async def run(
@ -332,7 +332,7 @@ class RemoteSandboxProvider:
mcp = FastMCP(
"Server",
transforms=[CodeMode(sandbox_provider=RemoteSandboxProvider())],
plugins=[CodeMode(sandbox_provider=RemoteSandboxProvider())],
)
```

14
examples/code_mode/server.py
View file

@ -1,4 +1,4 @@
"""Example: CodeMode transform — search and execute tools via code.
"""Example: CodeMode plugin — search and execute tools via code.
CodeMode replaces the entire tool catalog with two meta-tools: `search`
(keyword-based tool discovery) and `execute` (run Python code that chains
@ -13,9 +13,9 @@ Run with:
"""
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import CodeMode
from fastmcp.server.plugins.code_mode import CodeMode
mcp = FastMCP("CodeMode Demo")
mcp = FastMCP("CodeMode Demo", plugins=[CodeMode()])
@mcp.tool
@ -74,10 +74,10 @@ def read_file(path: str) -> str:
return f.read()
# CodeMode collapses all 8 tools into just `search` + `execute`.
# The LLM discovers tools via keyword search, then writes Python
# scripts that chain multiple tool calls in a single round-trip.
mcp.add_transform(CodeMode())
# CodeMode (registered at construction above) collapses all 8 tools
# into just `search` + `execute`. The LLM discovers tools via keyword
# search, then writes Python scripts that chain multiple tool calls in
# a single round-trip.
if __name__ == "__main__":

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

@ -1,567 +1,62 @@
import importlib
import json
from collections.abc import Awaitable, Callable, Sequence
from typing import TYPE_CHECKING, Annotated, Any, Literal, Protocol
"""Deprecation shim — code mode moved to `fastmcp.server.plugins.code_mode`.
if TYPE_CHECKING:
from pydantic_monty import ResourceLimits
The preferred API is now the `CodeMode` plugin:
from mcp.types import TextContent
from pydantic import Field
from fastmcp import FastMCP
from fastmcp.server.plugins.code_mode import CodeMode
from fastmcp.exceptions import NotFoundError
from fastmcp.server.context import Context
from fastmcp.server.plugins.tool_search.base import (
serialize_tools_for_output_json,
serialize_tools_for_output_markdown,
)
from fastmcp.server.plugins.tool_search.bm25 import BM25SearchTransform
from fastmcp.server.transforms import GetToolNext
from fastmcp.server.transforms.catalog import CatalogTransform
from fastmcp.tools.base import Tool, ToolResult
from fastmcp.utilities.async_utils import is_coroutine_function
from fastmcp.utilities.versions import VersionSpec
mcp = FastMCP("Server", plugins=[CodeMode()])
# ---------------------------------------------------------------------------
# Type aliases
# ---------------------------------------------------------------------------
For backcompat, this module keeps `CodeMode` bound to the **transform**
class (so existing `mcp.add_transform(CodeMode())` code keeps working).
The transform is also exported under its new canonical name,
`CodeModeTransform`. Sandbox providers, discovery-tool factories, and
related helpers re-export from the new location unchanged.
GetToolCatalog = Callable[[Context], Awaitable[Sequence[Tool]]]
"""Async callable that returns the auth-filtered tool catalog."""
SearchFn = Callable[[Sequence[Tool], str], Awaitable[Sequence[Tool]]]
"""Async callable that searches a tool sequence by query string."""
DiscoveryToolFactory = Callable[[GetToolCatalog], Tool]
"""Factory that receives catalog access and returns a synthetic Tool."""
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _ensure_async(fn: Callable[..., Any]) -> Callable[..., Any]:
if is_coroutine_function(fn):
return fn
async def wrapper(*args: Any, **kwargs: Any) -> Any:
return fn(*args, **kwargs)
return wrapper
def _unwrap_tool_result(result: ToolResult) -> dict[str, Any] | str:
"""Convert a ToolResult for use in the sandbox.
- Output schema present → structured_content dict (matches the schema)
- Otherwise → concatenated text content as a string
"""
if result.structured_content is not None:
return result.structured_content
parts: list[str] = []
for content in result.content:
if isinstance(content, TextContent):
parts.append(content.text)
else:
parts.append(str(content))
return "\n".join(parts)
# ---------------------------------------------------------------------------
# Sandbox providers
# ---------------------------------------------------------------------------
class SandboxProvider(Protocol):
"""Interface for executing LLM-generated Python code in a sandbox.
WARNING: The ``code`` parameter passed to ``run`` contains untrusted,
LLM-generated Python. Implementations MUST execute it in an isolated
sandbox — never with plain ``exec()``. Use ``MontySandboxProvider``
(backed by ``pydantic-monty``) for production workloads.
"""
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Callable[..., Any]] | None = None,
) -> Any: ...
class MontySandboxProvider:
"""Sandbox provider backed by `pydantic-monty`.
Args:
limits: Resource limits for sandbox execution. Supported keys:
``max_duration_secs`` (float), ``max_allocations`` (int),
``max_memory`` (int), ``max_recursion_depth`` (int),
``gc_interval`` (int). All are optional; omit a key to
leave that limit uncapped.
"""
def __init__(
self,
*,
limits: "ResourceLimits | None" = None,
) -> None:
self.limits = limits
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Callable[..., Any]] | None = None,
) -> Any:
try:
pydantic_monty = importlib.import_module("pydantic_monty")
except ModuleNotFoundError as exc:
raise ImportError(
"CodeMode requires pydantic-monty for the Monty sandbox provider. "
"Install it with `fastmcp[code-mode]` or pass a custom SandboxProvider."
) from exc
inputs = inputs or {}
async_functions = {
key: _ensure_async(value)
for key, value in (external_functions or {}).items()
}
monty = pydantic_monty.Monty(code, inputs=list(inputs))
return await monty.run_async(
inputs=inputs or None,
external_functions=async_functions or None,
limits=self.limits,
)
# ---------------------------------------------------------------------------
# Built-in discovery tools
# ---------------------------------------------------------------------------
ToolDetailLevel = Literal["brief", "detailed", "full"]
"""Detail level for discovery tool output.
- ``"brief"``: tool names and one-line descriptions
- ``"detailed"``: compact markdown with parameter names, types, and required markers
- ``"full"``: complete JSON schema
This path issues a `FastMCPDeprecationWarning` on import — a
`DeprecationWarning` subclass that fastmcp enables by default (plain
`DeprecationWarning` is suppressed by CPython's default filter, so
users wouldn't see the notice).
"""
def _render_tools(tools: Sequence[Tool], detail: ToolDetailLevel) -> str:
"""Render tools at the requested detail level.
The same detail value produces the same output format regardless of
which discovery tool calls this, so ``detail="detailed"`` on Search
gives identical formatting to ``detail="detailed"`` on GetSchemas.
"""
if not tools:
if detail == "full":
return json.dumps([], indent=2)
return "No tools matched the query."
if detail == "full":
return json.dumps(serialize_tools_for_output_json(tools), indent=2)
if detail == "detailed":
return serialize_tools_for_output_markdown(tools)
# brief
lines: list[str] = []
for tool in tools:
desc = f": {tool.description}" if tool.description else ""
lines.append(f"- {tool.name}{desc}")
return "\n".join(lines)
class Search:
"""Discovery tool factory that searches the catalog by query.
Args:
search_fn: Async callable ``(tools, query) -> matching_tools``.
Defaults to BM25 ranking.
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level for search results.
``"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__(
self,
*,
search_fn: SearchFn | None = None,
name: str = "search",
default_detail: ToolDetailLevel | None = None,
default_limit: int | None = None,
) -> None:
if search_fn is None:
_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"],
tags: Annotated[
list[str] | None,
"Filter to tools with any of these tags before searching",
] = None,
detail: Annotated[
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] # ty:ignore[invalid-parameter-default]
) -> str:
"""Search for available tools by query.
Returns matching tools ranked by relevance.
"""
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
real_tags = tag_set - {"untagged"}
tools = [
t
for t in tools
if (t.tags & real_tags) or (has_untagged and not t.tags)
]
results = await search_fn(tools, query)
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)
class GetSchemas:
"""Discovery tool factory that returns schemas for tools by name.
Args:
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level for schema results.
``"brief"`` returns tool names and descriptions only.
``"detailed"`` renders compact markdown with parameter names,
types, and required markers.
``"full"`` returns the complete JSON schema.
"""
def __init__(
self,
*,
name: str = "get_schema",
default_detail: ToolDetailLevel | None = None,
) -> None:
self._name = name
self._default_detail: ToolDetailLevel = default_detail or "detailed"
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
default_detail = self._default_detail
async def get_schema(
tools: Annotated[
list[str],
"List of tool names to get schemas for",
],
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] # ty:ignore[invalid-parameter-default]
) -> str:
"""Get parameter schemas for specific tools.
Use after searching to get the detail needed to call a tool.
"""
catalog = await get_catalog(ctx)
catalog_by_name = {t.name: t for t in catalog}
matched = [catalog_by_name[n] for n in tools if n in catalog_by_name]
not_found = [n for n in tools if n not in catalog_by_name]
if not matched and not_found:
return f"Tools not found: {', '.join(not_found)}"
if detail == "full":
data = serialize_tools_for_output_json(matched)
if not_found:
data.append({"not_found": not_found})
return json.dumps(data, indent=2)
result = _render_tools(matched, detail)
if not_found:
result += f"\n\nTools not found: {', '.join(not_found)}"
return result
return Tool.from_function(fn=get_schema, name=self._name)
class GetTags:
"""Discovery tool factory that lists tool tags from the catalog.
Reads ``tool.tags`` from the catalog and groups tools by tag. Tools
without tags appear under ``"untagged"``.
Args:
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level.
``"brief"`` returns tag names with tool counts.
``"full"`` lists all tools under each tag.
"""
def __init__(
self,
*,
name: str = "tags",
default_detail: Literal["brief", "full"] | None = None,
) -> None:
self._name = name
self._default_detail: Literal["brief", "full"] = default_detail or "brief"
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
default_detail = self._default_detail
async def tags(
detail: Annotated[
Literal["brief", "full"],
"Level of detail: 'brief' for tag names and counts, 'full' for tools listed under each tag",
] = default_detail,
ctx: Context = None, # type: ignore[assignment] # ty:ignore[invalid-parameter-default]
) -> str:
"""List available tool tags.
Use to browse available tools by tag before searching.
"""
catalog = await get_catalog(ctx)
by_tag: dict[str, list[Tool]] = {}
for tool in catalog:
if tool.tags:
for tag in tool.tags:
by_tag.setdefault(tag, []).append(tool)
else:
by_tag.setdefault("untagged", []).append(tool)
if not by_tag:
return "No tools available."
if detail == "brief":
lines = [
f"- {tag} ({len(tools)} tool{'s' if len(tools) != 1 else ''})"
for tag, tools in sorted(by_tag.items())
]
return "\n".join(lines)
blocks: list[str] = []
for tag, tools in sorted(by_tag.items()):
lines = [f"### {tag}"]
for tool in tools:
desc = f": {tool.description}" if tool.description else ""
lines.append(f"- {tool.name}{desc}")
blocks.append("\n".join(lines))
return "\n\n".join(blocks)
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] # ty:ignore[invalid-parameter-default]
) -> 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
# ---------------------------------------------------------------------------
def _default_discovery_tools() -> list[DiscoveryToolFactory]:
return [Search(), GetSchemas()]
class CodeMode(CatalogTransform):
"""Transform that collapses all tools into discovery + execute meta-tools.
Discovery tools are composable via the ``discovery_tools`` parameter.
Each is a callable that receives catalog access and returns a ``Tool``.
By default, ``Search`` and ``GetSchemas`` are included for
progressive disclosure: search finds candidates, get_schema retrieves
parameter details, and execute runs code.
The ``execute`` tool is always present and provides a sandboxed Python
environment with ``call_tool(name, params)`` in scope.
"""
def __init__(
self,
*,
sandbox_provider: SandboxProvider | None = None,
discovery_tools: list[DiscoveryToolFactory] | None = None,
execute_tool_name: str = "execute",
execute_description: str | None = None,
) -> None:
super().__init__()
self.execute_tool_name = execute_tool_name
self.execute_description = execute_description
self.sandbox_provider = sandbox_provider or MontySandboxProvider()
self._discovery_factories = (
discovery_tools
if discovery_tools is not None
else _default_discovery_tools()
)
self._built_discovery_tools: list[Tool] | None = None
self._cached_execute_tool: Tool | None = None
def _build_discovery_tools(self) -> list[Tool]:
if self._built_discovery_tools is None:
tools = [
factory(self.get_tool_catalog) for factory in self._discovery_factories
]
names = {t.name for t in tools}
if self.execute_tool_name in names:
raise ValueError(
f"Discovery tool name '{self.execute_tool_name}' "
f"collides with execute_tool_name."
)
if len(names) != len(tools):
raise ValueError("Discovery tools must have unique names.")
self._built_discovery_tools = tools
return self._built_discovery_tools
async def transform_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
return [*self._build_discovery_tools(), self._get_execute_tool()]
async def get_tool(
self,
name: str,
call_next: GetToolNext,
*,
version: VersionSpec | None = None,
) -> Tool | None:
for tool in self._build_discovery_tools():
if tool.name == name:
return tool
if name == self.execute_tool_name:
return self._get_execute_tool()
return await call_next(name, version=version)
def _build_execute_description(self) -> str:
if self.execute_description is not None:
return self.execute_description
return (
"Chain `await call_tool(...)` calls in one Python block; prefer returning the final answer from a single block.\n"
"Use `return` to produce output.\n"
"Only `call_tool(tool_name: str, params: dict) -> Any` is available in scope."
)
@staticmethod
def _find_tool(name: str, tools: Sequence[Tool]) -> Tool | None:
"""Find a tool by name from a pre-fetched list."""
for tool in tools:
if tool.name == name:
return tool
return None
def _get_execute_tool(self) -> Tool:
if self._cached_execute_tool is None:
self._cached_execute_tool = self._make_execute_tool()
return self._cached_execute_tool
def _make_execute_tool(self) -> Tool:
transform = self
async def execute(
code: Annotated[
str,
Field(
description=(
"Python async code to execute tool calls via call_tool(name, arguments)"
)
),
],
ctx: Context = None, # type: ignore[assignment] # ty:ignore[invalid-parameter-default]
) -> Any:
"""Execute tool calls using Python code."""
async def call_tool(tool_name: str, params: dict[str, Any]) -> Any:
backend_tools = await transform.get_tool_catalog(ctx)
tool = transform._find_tool(tool_name, backend_tools)
if tool is None:
raise NotFoundError(f"Unknown tool: {tool_name}")
result = await ctx.fastmcp.call_tool(tool.name, params)
return _unwrap_tool_result(result)
return await transform.sandbox_provider.run(
code,
external_functions={"call_tool": call_tool},
)
return Tool.from_function(
fn=execute,
name=self.execute_tool_name,
description=self._build_execute_description(),
)
import warnings
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.server.plugins.code_mode.discovery import (
DiscoveryToolFactory,
GetSchemas,
GetTags,
GetToolCatalog,
ListTools,
Search,
)
from fastmcp.server.plugins.code_mode.sandbox import (
MontySandboxProvider,
SandboxProvider,
)
from fastmcp.server.plugins.code_mode.transform import CodeModeTransform
# `CodeMode` at this old path stays bound to the transform class, so
# `mcp.add_transform(CodeMode(...))` keeps working. The new plugin class
# is at `fastmcp.server.plugins.code_mode.CodeMode`.
CodeMode = CodeModeTransform
warnings.warn(
"fastmcp.experimental.transforms.code_mode has moved to "
"fastmcp.server.plugins.code_mode. Prefer the CodeMode plugin: "
"`from fastmcp.server.plugins.code_mode import CodeMode` and pass "
"it via `plugins=[CodeMode(...)]`. At this old path, `CodeMode` "
"remains the transform class (also exported as `CodeModeTransform`) "
"for backcompat. The old import path will be removed in a future "
"release.",
FastMCPDeprecationWarning,
stacklevel=2,
)
__all__ = [
"CodeMode",
"CodeModeTransform",
"DiscoveryToolFactory",
"GetSchemas",
"GetTags",
"GetToolCatalog",

42
src/fastmcp/server/plugins/code_mode/__init__.py Normal file
View file

@ -0,0 +1,42 @@
"""Code-mode plugin — discovery + sandboxed Python execution in place of the tool catalog.
The `CodeMode` plugin is the public entry point:
from fastmcp import FastMCP
from fastmcp.server.plugins.code_mode import CodeMode
mcp = FastMCP("Server", plugins=[CodeMode()])
Discovery-tool factories (`Search`, `GetSchemas`, `GetTags`,
`ListTools`) and the sandbox-provider protocol (`SandboxProvider`,
`MontySandboxProvider`) are re-exported for custom composition. The
low-level `CodeModeTransform` lives in `.transform` for advanced users
who want to stack it directly with other transforms.
"""
from fastmcp.server.plugins.code_mode.discovery import (
DiscoveryToolFactory,
GetSchemas,
GetTags,
GetToolCatalog,
ListTools,
Search,
)
from fastmcp.server.plugins.code_mode.plugin import CodeMode, CodeModeConfig
from fastmcp.server.plugins.code_mode.sandbox import (
MontySandboxProvider,
SandboxProvider,
)
__all__ = [
"CodeMode",
"CodeModeConfig",
"DiscoveryToolFactory",
"GetSchemas",
"GetTags",
"GetToolCatalog",
"ListTools",
"MontySandboxProvider",
"SandboxProvider",
"Search",
]

323
src/fastmcp/server/plugins/code_mode/discovery.py Normal file
View file

@ -0,0 +1,323 @@
"""Discovery tool factories for the CodeMode plugin.
A discovery tool is a synthetic meta-tool the LLM uses to explore the real
tool catalog before calling anything. Each factory here is a callable
that receives catalog access (`GetToolCatalog`) and returns a ready-to-
publish `Tool`. They compose via the `discovery_tools` parameter on
`CodeMode`.
The four built-in factories cover the common discovery patterns:
* `Search` — query the catalog by text (BM25 by default).
* `GetSchemas` — fetch parameter schemas for a named list of tools.
* `GetTags` — browse the catalog grouped by tag.
* `ListTools` — dump every tool at a configurable detail level.
A typical progressive-disclosure setup pairs `Search` with `GetSchemas`:
the LLM searches to find candidates, then fetches schemas only for the
tools it actually plans to call.
"""
from __future__ import annotations
import json
from collections.abc import Awaitable, Callable, Sequence
from typing import Annotated, Literal
from fastmcp.server.context import Context
from fastmcp.server.plugins.tool_search.base import (
serialize_tools_for_output_json,
serialize_tools_for_output_markdown,
)
from fastmcp.tools.base import Tool
GetToolCatalog = Callable[[Context], Awaitable[Sequence[Tool]]]
"""Async callable that returns the auth-filtered tool catalog."""
SearchFn = Callable[[Sequence[Tool], str], Awaitable[Sequence[Tool]]]
"""Async callable that searches a tool sequence by query string."""
DiscoveryToolFactory = Callable[[GetToolCatalog], Tool]
"""Factory that receives catalog access and returns a synthetic Tool."""
ToolDetailLevel = Literal["brief", "detailed", "full"]
"""Detail level for discovery tool output.
- `"brief"`: tool names and one-line descriptions
- `"detailed"`: compact markdown with parameter names, types, and required markers
- `"full"`: complete JSON schema
"""
def _render_tools(tools: Sequence[Tool], detail: ToolDetailLevel) -> str:
"""Render tools at the requested detail level.
The same detail value produces the same output format regardless of
which discovery tool calls this, so `detail="detailed"` on Search
gives identical formatting to `detail="detailed"` on GetSchemas.
"""
if not tools:
if detail == "full":
return json.dumps([], indent=2)
return "No tools matched the query."
if detail == "full":
return json.dumps(serialize_tools_for_output_json(tools), indent=2)
if detail == "detailed":
return serialize_tools_for_output_markdown(tools)
# brief
lines: list[str] = []
for tool in tools:
desc = f": {tool.description}" if tool.description else ""
lines.append(f"- {tool.name}{desc}")
return "\n".join(lines)
class Search:
"""Discovery tool factory that searches the catalog by query.
Args:
search_fn: Async callable `(tools, query) -> matching_tools`.
Defaults to BM25 ranking.
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level for search results.
`"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__(
self,
*,
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.plugins.tool_search.bm25 import 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"],
tags: Annotated[
list[str] | None,
"Filter to tools with any of these tags before searching",
] = None,
detail: Annotated[
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] # ty:ignore[invalid-parameter-default]
) -> str:
"""Search for available tools by query.
Returns matching tools ranked by relevance.
"""
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
real_tags = tag_set - {"untagged"}
tools = [
t
for t in tools
if (t.tags & real_tags) or (has_untagged and not t.tags)
]
results = await search_fn(tools, query)
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)
class GetSchemas:
"""Discovery tool factory that returns schemas for tools by name.
Args:
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level for schema results.
`"brief"` returns tool names and descriptions only.
`"detailed"` renders compact markdown with parameter names,
types, and required markers.
`"full"` returns the complete JSON schema.
"""
def __init__(
self,
*,
name: str = "get_schema",
default_detail: ToolDetailLevel | None = None,
) -> None:
self._name = name
self._default_detail: ToolDetailLevel = default_detail or "detailed"
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
default_detail = self._default_detail
async def get_schema(
tools: Annotated[
list[str],
"List of tool names to get schemas for",
],
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] # ty:ignore[invalid-parameter-default]
) -> str:
"""Get parameter schemas for specific tools.
Use after searching to get the detail needed to call a tool.
"""
catalog = await get_catalog(ctx)
catalog_by_name = {t.name: t for t in catalog}
matched = [catalog_by_name[n] for n in tools if n in catalog_by_name]
not_found = [n for n in tools if n not in catalog_by_name]
if not matched and not_found:
return f"Tools not found: {', '.join(not_found)}"
if detail == "full":
data = serialize_tools_for_output_json(matched)
if not_found:
data.append({"not_found": not_found})
return json.dumps(data, indent=2)
result = _render_tools(matched, detail)
if not_found:
result += f"\n\nTools not found: {', '.join(not_found)}"
return result
return Tool.from_function(fn=get_schema, name=self._name)
class GetTags:
"""Discovery tool factory that lists tool tags from the catalog.
Reads `tool.tags` from the catalog and groups tools by tag. Tools
without tags appear under `"untagged"`.
Args:
name: Name of the synthetic tool exposed to the LLM.
default_detail: Default detail level.
`"brief"` returns tag names with tool counts.
`"full"` lists all tools under each tag.
"""
def __init__(
self,
*,
name: str = "tags",
default_detail: Literal["brief", "full"] | None = None,
) -> None:
self._name = name
self._default_detail: Literal["brief", "full"] = default_detail or "brief"
def __call__(self, get_catalog: GetToolCatalog) -> Tool:
default_detail = self._default_detail
async def tags(
detail: Annotated[
Literal["brief", "full"],
"Level of detail: 'brief' for tag names and counts, 'full' for tools listed under each tag",
] = default_detail,
ctx: Context = None, # type: ignore[assignment] # ty:ignore[invalid-parameter-default]
) -> str:
"""List available tool tags.
Use to browse available tools by tag before searching.
"""
catalog = await get_catalog(ctx)
by_tag: dict[str, list[Tool]] = {}
for tool in catalog:
if tool.tags:
for tag in tool.tags:
by_tag.setdefault(tag, []).append(tool)
else:
by_tag.setdefault("untagged", []).append(tool)
if not by_tag:
return "No tools available."
if detail == "brief":
lines = [
f"- {tag} ({len(tools)} tool{'s' if len(tools) != 1 else ''})"
for tag, tools in sorted(by_tag.items())
]
return "\n".join(lines)
blocks: list[str] = []
for tag, tools in sorted(by_tag.items()):
lines = [f"### {tag}"]
for tool in tools:
desc = f": {tool.description}" if tool.description else ""
lines.append(f"- {tool.name}{desc}")
blocks.append("\n".join(lines))
return "\n\n".join(blocks)
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] # ty:ignore[invalid-parameter-default]
) -> 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)

129
src/fastmcp/server/plugins/code_mode/plugin.py Normal file
View file

@ -0,0 +1,129 @@
"""CodeMode plugin: tool execution via LLM-generated code.
`CodeMode` replaces the entire tool catalog with two classes of
meta-tool — discovery tools (search, get_schema, etc.) and a single
`execute` tool that runs LLM-generated Python in a sandbox. The model
discovers what's available on demand and chains calls inside one
sandboxed code block, which dramatically cuts round-trips and context
for servers with many tools.
"""
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict
from fastmcp.server.plugins.base import Plugin
from fastmcp.server.plugins.code_mode.discovery import DiscoveryToolFactory
from fastmcp.server.plugins.code_mode.sandbox import (
MontySandboxProvider,
SandboxProvider,
)
from fastmcp.server.plugins.code_mode.transform import CodeModeTransform
from fastmcp.server.transforms import Transform
class CodeModeConfig(BaseModel):
"""Config model for the `CodeMode` plugin.
Only covers JSON-serializable settings — the sandbox provider and
discovery-tool factories are passed through `CodeMode.__init__`
directly because they're real Python objects.
"""
model_config = ConfigDict(extra="forbid")
sandbox: Literal["monty"] = "monty"
"""Built-in sandbox provider to use. `"monty"` uses
`pydantic-monty`. For a custom provider, pass `sandbox_provider=...`
to `CodeMode.__init__` instead."""
sandbox_limits: dict[str, Any] | None = None
"""Resource limits for the default Monty sandbox. Keys:
`max_duration_secs`, `max_allocations`, `max_memory`,
`max_recursion_depth`, `gc_interval`. All optional."""
execute_tool_name: str = "execute"
"""Name of the generated execute tool."""
execute_description: str | None = None
"""Override the default description of the execute tool. `None`
keeps the built-in guidance."""
class CodeMode(Plugin[CodeModeConfig]):
"""Collapse the tool catalog behind discovery + code-execution meta-tools.
Users write a CodeMode-enabled server exactly like a normal server;
the plugin takes care of hiding the real tools and exposing search
/ get_schema / execute in their place.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.plugins.code_mode import CodeMode
mcp = FastMCP("Server", plugins=[CodeMode()])
```
For a custom sandbox or custom discovery-tool set, pass Python
objects through `__init__`:
```python
from fastmcp.server.plugins.code_mode import (
CodeMode,
CodeModeConfig,
GetSchemas,
ListTools,
)
mcp = FastMCP(
"Server",
plugins=[
CodeMode(
CodeModeConfig(execute_tool_name="run"),
sandbox_provider=my_custom_sandbox,
discovery_tools=[ListTools(), GetSchemas()],
)
],
)
```
"""
# `meta` is auto-derived (name="code-mode", version=None) — the right
# answer for a bundled first-party plugin. Declare `meta` explicitly
# (or use `PluginMeta.from_package(...)`) if published separately.
def __init__(
self,
config: CodeModeConfig | dict[str, Any] | None = None,
*,
sandbox_provider: SandboxProvider | None = None,
discovery_tools: list[DiscoveryToolFactory] | None = None,
) -> None:
super().__init__(config)
self._sandbox_override = sandbox_provider
self._discovery_override = discovery_tools
def transforms(self) -> list[Transform]:
sandbox = self._sandbox_override or self._build_default_sandbox()
return [
CodeModeTransform(
sandbox_provider=sandbox,
discovery_tools=self._discovery_override,
execute_tool_name=self.config.execute_tool_name,
execute_description=self.config.execute_description,
)
]
def _build_default_sandbox(self) -> SandboxProvider:
limits_dict = self.config.sandbox_limits
if limits_dict is None:
return MontySandboxProvider()
# Defer the import so Monty is only a hard dependency when
# `sandbox_limits` is actually configured.
from pydantic_monty import ResourceLimits
return MontySandboxProvider(limits=ResourceLimits(**limits_dict))

94
src/fastmcp/server/plugins/code_mode/sandbox.py Normal file
View file

@ -0,0 +1,94 @@
"""Sandbox providers for the CodeMode plugin.
A `SandboxProvider` is the component that actually executes LLM-generated
Python code. The default `MontySandboxProvider` delegates to
`pydantic-monty` for isolated execution; alternative providers can plug in
any other sandbox (remote process, WASM, a containerized worker, etc.)
by implementing the `SandboxProvider` protocol.
"""
from __future__ import annotations
import importlib
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, Protocol
from fastmcp.utilities.async_utils import is_coroutine_function
if TYPE_CHECKING:
from pydantic_monty import ResourceLimits
def _ensure_async(fn: Callable[..., Any]) -> Callable[..., Any]:
if is_coroutine_function(fn):
return fn
async def wrapper(*args: Any, **kwargs: Any) -> Any:
return fn(*args, **kwargs)
return wrapper
class SandboxProvider(Protocol):
"""Interface for executing LLM-generated Python code in a sandbox.
WARNING: The `code` parameter passed to `run` contains untrusted,
LLM-generated Python. Implementations MUST execute it in an isolated
sandbox — never with plain `exec()`. Use `MontySandboxProvider`
(backed by `pydantic-monty`) for production workloads.
"""
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Callable[..., Any]] | None = None,
) -> Any: ...
class MontySandboxProvider:
"""Sandbox provider backed by `pydantic-monty`.
Args:
limits: Resource limits for sandbox execution. Supported keys:
`max_duration_secs` (float), `max_allocations` (int),
`max_memory` (int), `max_recursion_depth` (int),
`gc_interval` (int). All are optional; omit a key to leave
that limit uncapped.
"""
def __init__(
self,
*,
limits: ResourceLimits | None = None,
) -> None:
self.limits = limits
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Callable[..., Any]] | None = None,
) -> Any:
try:
pydantic_monty = importlib.import_module("pydantic_monty")
except ModuleNotFoundError as exc:
raise ImportError(
"CodeMode requires pydantic-monty for the Monty sandbox provider. "
"Install it with `fastmcp[code-mode]` or pass a custom SandboxProvider."
) from exc
inputs = inputs or {}
async_functions = {
key: _ensure_async(value)
for key, value in (external_functions or {}).items()
}
monty = pydantic_monty.Monty(code, inputs=list(inputs))
return await monty.run_async(
inputs=inputs or None,
external_functions=async_functions or None,
limits=self.limits,
)

186
src/fastmcp/server/plugins/code_mode/transform.py Normal file
View file

@ -0,0 +1,186 @@
"""Low-level transform that powers the CodeMode plugin.
`CodeModeTransform` replaces the tool catalog with two classes of
meta-tool: configurable **discovery tools** (search, get_schema, etc.)
that let the LLM explore what's available, and a single **execute tool**
that runs LLM-generated Python in a sandbox with `call_tool(...)`
available in scope.
Most users should configure CodeMode through the `CodeMode` plugin
(`fastmcp.server.plugins.code_mode`). The transform is exposed for
advanced composition — users who want to stack it with other transforms
directly or embed it in a custom plugin.
"""
from __future__ import annotations
from collections.abc import Sequence
from typing import Annotated, Any
from mcp.types import TextContent
from pydantic import Field
from fastmcp.exceptions import NotFoundError
from fastmcp.server.context import Context
from fastmcp.server.plugins.code_mode.discovery import (
DiscoveryToolFactory,
GetSchemas,
Search,
)
from fastmcp.server.plugins.code_mode.sandbox import (
MontySandboxProvider,
SandboxProvider,
)
from fastmcp.server.transforms import GetToolNext
from fastmcp.server.transforms.catalog import CatalogTransform
from fastmcp.tools.base import Tool, ToolResult
from fastmcp.utilities.versions import VersionSpec
def _unwrap_tool_result(result: ToolResult) -> dict[str, Any] | str:
"""Convert a ToolResult for use in the sandbox.
- Output schema present → structured_content dict (matches the schema)
- Otherwise → concatenated text content as a string
"""
if result.structured_content is not None:
return result.structured_content
parts: list[str] = []
for content in result.content:
if isinstance(content, TextContent):
parts.append(content.text)
else:
parts.append(str(content))
return "\n".join(parts)
def _default_discovery_tools() -> list[DiscoveryToolFactory]:
return [Search(), GetSchemas()]
class CodeModeTransform(CatalogTransform):
"""Transform that collapses all tools into discovery + execute meta-tools.
Discovery tools are composable via the `discovery_tools` parameter.
Each is a callable that receives catalog access and returns a `Tool`.
By default, `Search` and `GetSchemas` are included for progressive
disclosure: search finds candidates, get_schema retrieves parameter
details, and execute runs code.
The `execute` tool is always present and provides a sandboxed Python
environment with `call_tool(name, params)` in scope.
"""
def __init__(
self,
*,
sandbox_provider: SandboxProvider | None = None,
discovery_tools: list[DiscoveryToolFactory] | None = None,
execute_tool_name: str = "execute",
execute_description: str | None = None,
) -> None:
super().__init__()
self.execute_tool_name = execute_tool_name
self.execute_description = execute_description
self.sandbox_provider = sandbox_provider or MontySandboxProvider()
self._discovery_factories = (
discovery_tools
if discovery_tools is not None
else _default_discovery_tools()
)
self._built_discovery_tools: list[Tool] | None = None
self._cached_execute_tool: Tool | None = None
def _build_discovery_tools(self) -> list[Tool]:
if self._built_discovery_tools is None:
tools = [
factory(self.get_tool_catalog) for factory in self._discovery_factories
]
names = {t.name for t in tools}
if self.execute_tool_name in names:
raise ValueError(
f"Discovery tool name '{self.execute_tool_name}' "
f"collides with execute_tool_name."
)
if len(names) != len(tools):
raise ValueError("Discovery tools must have unique names.")
self._built_discovery_tools = tools
return self._built_discovery_tools
async def transform_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
return [*self._build_discovery_tools(), self._get_execute_tool()]
async def get_tool(
self,
name: str,
call_next: GetToolNext,
*,
version: VersionSpec | None = None,
) -> Tool | None:
for tool in self._build_discovery_tools():
if tool.name == name:
return tool
if name == self.execute_tool_name:
return self._get_execute_tool()
return await call_next(name, version=version)
def _build_execute_description(self) -> str:
if self.execute_description is not None:
return self.execute_description
return (
"Chain `await call_tool(...)` calls in one Python block; prefer returning the final answer from a single block.\n"
"Use `return` to produce output.\n"
"Only `call_tool(tool_name: str, params: dict) -> Any` is available in scope."
)
@staticmethod
def _find_tool(name: str, tools: Sequence[Tool]) -> Tool | None:
"""Find a tool by name from a pre-fetched list."""
for tool in tools:
if tool.name == name:
return tool
return None
def _get_execute_tool(self) -> Tool:
if self._cached_execute_tool is None:
self._cached_execute_tool = self._make_execute_tool()
return self._cached_execute_tool
def _make_execute_tool(self) -> Tool:
transform = self
async def execute(
code: Annotated[
str,
Field(
description=(
"Python async code to execute tool calls via call_tool(name, arguments)"
)
),
],
ctx: Context = None, # type: ignore[assignment] # ty:ignore[invalid-parameter-default]
) -> Any:
"""Execute tool calls using Python code."""
async def call_tool(tool_name: str, params: dict[str, Any]) -> Any:
backend_tools = await transform.get_tool_catalog(ctx)
tool = transform._find_tool(tool_name, backend_tools)
if tool is None:
raise NotFoundError(f"Unknown tool: {tool_name}")
result = await ctx.fastmcp.call_tool(tool.name, params)
return _unwrap_tool_result(result)
return await transform.sandbox_provider.run(
code,
external_functions={"call_tool": call_tool},
)
return Tool.from_function(
fn=execute,
name=self.execute_tool_name,
description=self._build_execute_description(),
)

66
tests/experimental/transforms/test_code_mode.py → tests/server/plugins/test_code_mode.py
View file

@ -7,15 +7,15 @@ from mcp.types import ImageContent, TextContent
from fastmcp import Client, FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.experimental.transforms.code_mode import (
CodeMode,
from fastmcp.server.context import Context
from fastmcp.server.plugins.code_mode import (
GetSchemas,
GetToolCatalog,
MontySandboxProvider,
Search,
_ensure_async,
)
from fastmcp.server.context import Context
from fastmcp.server.plugins.code_mode.sandbox import _ensure_async
from fastmcp.server.plugins.code_mode.transform import CodeModeTransform
from fastmcp.tools.base import Tool, ToolResult
@ -105,7 +105,7 @@ async def test_code_mode_default_tools() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
listed_tools = await mcp.list_tools(run_middleware=False)
assert {tool.name for tool in listed_tools} == {"search", "get_schema", "execute"}
@ -125,7 +125,7 @@ async def test_code_mode_search_returns_lightweight_results() -> None:
"""Say hello to someone."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "square number"})
text = _unwrap_string_result(result)
@ -144,7 +144,7 @@ async def test_code_mode_get_schema_brief() -> None:
"""Compute the square of a number."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "get_schema", {"tools": ["square"], "detail": "brief"}
@ -165,7 +165,7 @@ async def test_code_mode_get_schema_detailed() -> None:
"""Compute the square of a number."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "get_schema", {"tools": ["square"], "detail": "detailed"}
@ -186,7 +186,7 @@ async def test_code_mode_get_schema_full() -> None:
"""Compute the square of a number."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": ["square"], "detail": "full"})
text = _unwrap_string_result(result)
@ -205,7 +205,7 @@ async def test_code_mode_get_schema_default_is_detailed() -> None:
"""Compute the square of a number."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": ["square"]})
text = _unwrap_string_result(result)
@ -221,7 +221,7 @@ async def test_code_mode_get_schema_not_found() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": ["nonexistent"]})
text = _unwrap_string_result(result)
@ -238,7 +238,7 @@ async def test_code_mode_get_schema_partial_match() -> None:
"""Compute the square."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": ["square", "nonexistent"]})
text = _unwrap_string_result(result)
@ -254,7 +254,7 @@ async def test_code_mode_execute_works() -> None:
def add(x: int, y: int) -> int:
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "execute", {"code": "return await call_tool('add', {'x': 2, 'y': 3})"}
@ -275,7 +275,7 @@ async def test_code_mode_custom_execute_name() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
sandbox_provider=_UnsafeTestSandboxProvider(),
execute_tool_name="run_code",
)
@ -295,7 +295,7 @@ async def test_code_mode_custom_execute_description() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
sandbox_provider=_UnsafeTestSandboxProvider(),
execute_description="Custom execute description",
)
@ -313,7 +313,7 @@ async def test_code_mode_default_execute_description() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
listed = await mcp.list_tools(run_middleware=False)
by_name = {t.name: t for t in listed}
@ -341,7 +341,7 @@ async def test_code_mode_no_discovery_tools() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -371,7 +371,7 @@ async def test_code_mode_custom_discovery_tool_function() -> None:
return Tool.from_function(fn=list_tools, name="list_all")
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[list_all],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -394,7 +394,7 @@ async def test_code_mode_search_detailed() -> None:
"""Compute the square."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "square", "detail": "detailed"})
text = _unwrap_string_result(result)
@ -414,7 +414,7 @@ async def test_code_mode_search_tool_full_detail() -> None:
return x * x
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[Search(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -437,7 +437,7 @@ async def test_code_mode_custom_search_tool_name() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[
Search(name="find"),
GetSchemas(name="describe"),
@ -452,7 +452,7 @@ async def test_code_mode_custom_search_tool_name() -> None:
def test_code_mode_rejects_discovery_execute_name_collision() -> None:
"""CodeMode raises ValueError when a discovery tool collides with execute."""
cm = CodeMode(
cm = CodeModeTransform(
discovery_tools=[Search(name="execute")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -462,7 +462,7 @@ def test_code_mode_rejects_discovery_execute_name_collision() -> None:
def test_code_mode_rejects_duplicate_discovery_names() -> None:
"""CodeMode raises ValueError when discovery tools have duplicate names."""
cm = CodeMode(
cm = CodeModeTransform(
discovery_tools=[Search(name="search"), Search(name="search")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -483,7 +483,7 @@ async def test_code_mode_execute_respects_disabled_tool_visibility() -> None:
return "nope"
mcp.disable(names={"secret"}, components={"tool"})
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
with pytest.raises(ToolError, match=r"Unknown tool"):
await _run_tool(
@ -500,7 +500,7 @@ async def test_code_mode_search_respects_disabled_tool_visibility() -> None:
return "nope"
mcp.disable(names={"secret"}, components={"tool"})
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "secret"})
text = _unwrap_string_result(result)
@ -521,7 +521,7 @@ async def test_code_mode_execute_sees_mid_run_visibility_changes() -> None:
return "secret-ok"
mcp.disable(names={"secret"}, components={"tool"})
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
async with Client(mcp) as client:
result = await client.call_tool(
@ -540,7 +540,7 @@ async def test_code_mode_execute_respects_tool_auth() -> None:
def protected() -> str:
return "nope"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
with pytest.raises(ToolError, match=r"Unknown tool"):
await _run_tool(
@ -556,7 +556,7 @@ async def test_code_mode_search_respects_tool_auth() -> None:
"""A protected tool."""
return "nope"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "protected"})
text = _unwrap_string_result(result)
@ -575,7 +575,7 @@ async def test_code_mode_shadows_colliding_tool_names() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
tools = await mcp.list_tools(run_middleware=False)
tool_names = {t.name for t in tools}
@ -600,7 +600,7 @@ async def test_code_mode_get_tool_returns_meta_tools_and_passes_through() -> Non
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
search_tool = await mcp.get_tool("search")
assert search_tool is not None
@ -631,7 +631,7 @@ async def test_code_mode_execute_non_text_content_stringified() -> None:
def image_tool() -> ImageContent:
return ImageContent(type="image", data="base64data", mimeType="image/png")
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "execute", {"code": "return await call_tool('image_tool', {})"}
@ -653,7 +653,7 @@ async def test_code_mode_execute_multi_tool_chaining() -> None:
def add_one(x: int) -> int:
return x + 1
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp,
@ -676,7 +676,7 @@ async def test_code_mode_sandbox_error_surfaces_as_tool_error() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
with pytest.raises(ToolError):
await _run_tool(mcp, "execute", {"code": "raise ValueError('boom')"})

50
tests/experimental/transforms/test_code_mode_discovery.py → tests/server/plugins/test_code_mode_discovery.py
View file

@ -4,13 +4,13 @@ from typing import Any
from mcp.types import TextContent
from fastmcp import FastMCP
from fastmcp.experimental.transforms.code_mode import (
CodeMode,
from fastmcp.server.plugins.code_mode import (
GetTags,
ListTools,
Search,
_ensure_async,
)
from fastmcp.server.plugins.code_mode.sandbox import _ensure_async
from fastmcp.server.plugins.code_mode.transform import CodeModeTransform
from fastmcp.tools.base import ToolResult
@ -104,7 +104,7 @@ async def test_categories_brief_shows_tag_counts() -> None:
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -130,7 +130,7 @@ async def test_categories_full_lists_tools_per_tag() -> None:
return f"Hello, {name}!"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -156,7 +156,7 @@ async def test_categories_includes_untagged() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -176,7 +176,7 @@ async def test_categories_tool_in_multiple_tags() -> None:
return x + y
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags(default_detail="full")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -200,7 +200,7 @@ async def test_categories_detail_override_per_call() -> None:
return x + y
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags()], # default_detail="brief"
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -219,7 +219,7 @@ async def test_get_tags_empty_catalog() -> None:
mcp.disable(names={"ping"}, components={"tool"})
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[GetTags()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -248,7 +248,7 @@ async def test_search_with_tags_filter() -> None:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello", "tags": ["math"]})
text = _unwrap_string_result(result)
@ -264,7 +264,7 @@ async def test_search_with_tags_filter_no_matches() -> None:
"""Add two numbers."""
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add", "tags": ["nonexistent"]})
text = _unwrap_string_result(result)
@ -285,7 +285,7 @@ async def test_search_without_tags_returns_all() -> None:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add hello"})
text = _unwrap_string_result(result)
@ -307,7 +307,7 @@ async def test_search_with_untagged_filter() -> None:
"""Ping."""
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "ping add", "tags": ["untagged"]})
text = _unwrap_string_result(result)
@ -325,7 +325,7 @@ async def test_search_default_detail_detailed_skips_get_schema() -> None:
return x * x
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[Search(default_detail="detailed")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -346,7 +346,7 @@ async def test_search_full_detail_empty_results_returns_json() -> None:
def add(x: int, y: int) -> int:
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp,
@ -366,7 +366,7 @@ async def test_get_schema_empty_tools_list() -> None:
def ping() -> str:
return "pong"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "get_schema", {"tools": []})
text = _unwrap_string_result(result)
@ -382,7 +382,7 @@ async def test_get_schema_full_partial_match_returns_valid_json() -> None:
"""Compute the square."""
return x * x
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(
mcp, "get_schema", {"tools": ["square", "nonexistent"], "detail": "full"}
@ -418,7 +418,7 @@ async def test_search_shows_catalog_size_when_results_are_subset() -> None:
"""Say hello."""
return f"Hello, {name}!"
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add numbers"})
text = _unwrap_string_result(result)
@ -435,7 +435,7 @@ async def test_search_omits_annotation_when_all_tools_returned() -> None:
"""Add two numbers."""
return x + y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "add numbers"})
text = _unwrap_string_result(result)
@ -466,7 +466,7 @@ async def test_search_limit_caps_results() -> None:
"""Multiply numbers."""
return x * y
mcp.add_transform(CodeMode(sandbox_provider=_UnsafeTestSandboxProvider()))
mcp.add_transform(CodeModeTransform(sandbox_provider=_UnsafeTestSandboxProvider()))
result = await _run_tool(mcp, "search", {"query": "numbers", "limit": 1})
text = _unwrap_string_result(result)
@ -496,7 +496,7 @@ async def test_search_default_limit_from_constructor() -> None:
return "c"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[Search(default_limit=2)],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -527,7 +527,7 @@ async def test_list_tools_brief() -> None:
return x * y
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -552,7 +552,7 @@ async def test_list_tools_detailed() -> None:
return x * x
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[ListTools(default_detail="detailed")],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -575,7 +575,7 @@ async def test_list_tools_full_returns_json() -> None:
return "pong"
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)
@ -593,7 +593,7 @@ async def test_list_tools_empty_catalog() -> None:
mcp = FastMCP("ListTools Empty")
mcp.add_transform(
CodeMode(
CodeModeTransform(
discovery_tools=[ListTools()],
sandbox_provider=_UnsafeTestSandboxProvider(),
)

105
tests/server/plugins/test_code_mode_plugin.py Normal file
View file

@ -0,0 +1,105 @@
"""Tests for the CodeMode plugin wrapper.
Transform behavior (what `CodeModeTransform` does to the catalog, how
discovery tools render, sandbox execution, etc.) is covered by
`test_code_mode.py` and `test_code_mode_discovery.py`. This file only
covers the plugin layer itself — config validation, meta derivation,
dict-config coercion, and the deprecation shim at the old import path.
"""
from __future__ import annotations
import warnings
from typing import Any
import pytest
from pydantic import ValidationError
from fastmcp import FastMCP
from fastmcp.server.plugins.code_mode import CodeMode, CodeModeConfig
class _NoopSandbox:
async def run(
self,
code: str,
*,
inputs: dict[str, Any] | None = None,
external_functions: dict[str, Any] | None = None,
) -> Any:
return None
class TestCodeModeConfig:
def test_config_generic_binding(self):
"""`Plugin[CodeModeConfig]` binds CodeModeConfig as the validated config type."""
assert CodeMode._config_cls is CodeModeConfig
def test_dict_config_accepted(self):
"""Dict config works for loading from JSON/YAML."""
plugin = CodeMode({"execute_tool_name": "go"})
assert plugin.config.execute_tool_name == "go"
def test_unknown_sandbox_rejected(self):
with pytest.raises((ValidationError, Exception), match="sandbox"):
CodeModeConfig(sandbox="docker") # ty: ignore[invalid-argument-type]
def test_unknown_config_key_rejected(self):
with pytest.raises((ValidationError, Exception), match="forbid|extra"):
CodeModeConfig(not_a_real_option=True) # ty: ignore[unknown-argument]
def test_default_meta(self):
"""CodeMode uses Plugin's auto-derived meta: kebab-cased class
name, no independent version (bundled first-party plugin)."""
assert CodeMode.meta.name == "code-mode"
assert CodeMode.meta.version is None
class TestDeprecationShim:
"""The old `fastmcp.experimental.transforms.code_mode` path still works but warns."""
def test_old_package_import_emits_deprecation_warning(self):
import importlib
import sys
from fastmcp.exceptions import FastMCPDeprecationWarning
sys.modules.pop("fastmcp.experimental.transforms.code_mode", None)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
importlib.import_module("fastmcp.experimental.transforms.code_mode")
fastmcp_deprecations = [
w for w in caught if issubclass(w.category, FastMCPDeprecationWarning)
]
assert any(
"plugins.code_mode" in str(w.message) for w in fastmcp_deprecations
), (
f"expected FastMCPDeprecationWarning pointing at plugins.code_mode, "
f"got {[(w.category.__name__, str(w.message)) for w in caught]}"
)
async def test_legacy_add_transform_pattern_still_works(self):
"""End-to-end: old `add_transform(CodeMode(...))` code keeps
working. The point of the shim is that this doesn't break — the
identity-check test alone wouldn't catch a regression where
`CodeMode` at the old path drifted to the plugin class."""
from fastmcp.exceptions import FastMCPDeprecationWarning
with warnings.catch_warnings():
warnings.simplefilter("ignore", FastMCPDeprecationWarning)
from fastmcp.experimental.transforms.code_mode import (
CodeMode as OldCodeMode,
)
mcp = FastMCP("legacy")
@mcp.tool
def ping() -> str:
return "pong"
mcp.add_transform(OldCodeMode(sandbox_provider=_NoopSandbox()))
tools = await mcp.list_tools(run_middleware=False)
assert {t.name for t in tools} == {"search", "get_schema", "execute"}

0
tests/experimental/transforms/test_code_mode_serialization.py → tests/server/plugins/test_code_mode_serialization.py
View file