Skip to content

Index

Tool

Bases: BaseModel

Internal tool registration info.

Source code in src/mcp/server/mcpserver/tools/base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class Tool(BaseModel):
    """Internal tool registration info."""

    fn: Callable[..., Any] = Field(exclude=True)
    name: str = Field(description="Name of the tool")
    title: str | None = Field(None, description="Human-readable title of the tool")
    description: str = Field(description="Description of what the tool does")
    parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
    fn_metadata: FuncMetadata = Field(
        description="Metadata about the function including a pydantic model for tool arguments"
    )
    is_async: bool = Field(description="Whether the tool is async")
    context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
    annotations: ToolAnnotations | None = Field(None, description="Optional annotations for the tool")
    icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this tool")
    meta: dict[str, Any] | None = Field(default=None, description="Optional metadata for this tool")

    @cached_property
    def output_schema(self) -> dict[str, Any] | None:
        return self.fn_metadata.output_schema

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., Any],
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        context_kwarg: str | None = None,
        annotations: ToolAnnotations | None = None,
        icons: list[Icon] | None = None,
        meta: dict[str, Any] | None = None,
        structured_output: bool | None = None,
    ) -> Tool:
        """Create a Tool from a function."""
        func_name = name or fn.__name__

        validate_and_warn_tool_name(func_name)

        if func_name == "<lambda>":
            raise ValueError("You must provide a name for lambda functions")

        func_doc = description or fn.__doc__ or ""
        is_async = is_async_callable(fn)

        if context_kwarg is None:  # pragma: no branch
            context_kwarg = find_context_parameter(fn)

        func_arg_metadata = func_metadata(
            fn,
            skip_names=[context_kwarg] if context_kwarg is not None else [],
            structured_output=structured_output,
        )
        parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)

        return cls(
            fn=fn,
            name=func_name,
            title=title,
            description=func_doc,
            parameters=parameters,
            fn_metadata=func_arg_metadata,
            is_async=is_async,
            context_kwarg=context_kwarg,
            annotations=annotations,
            icons=icons,
            meta=meta,
        )

    async def run(
        self,
        arguments: dict[str, Any],
        context: Context[LifespanContextT, RequestT],
        convert_result: bool = False,
    ) -> Any:
        """Run the tool with arguments.

        Raises:
            ToolError: If the tool function raises during execution.
        """
        try:
            result = await self.fn_metadata.call_fn_with_arg_validation(
                self.fn,
                self.is_async,
                arguments,
                {self.context_kwarg: context} if self.context_kwarg is not None else None,
            )

            if convert_result:
                result = self.fn_metadata.convert_result(result)

            return result
        except UrlElicitationRequiredError:
            # Re-raise UrlElicitationRequiredError so it can be properly handled
            # as an MCP error response with code -32042
            raise
        except Exception as e:
            raise ToolError(f"Error executing tool {self.name}: {e}") from e

from_function classmethod 

from_function(
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    context_kwarg: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Tool

Create a Tool from a function.

Source code in src/mcp/server/mcpserver/tools/base.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def from_function(
    cls,
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    context_kwarg: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Tool:
    """Create a Tool from a function."""
    func_name = name or fn.__name__

    validate_and_warn_tool_name(func_name)

    if func_name == "<lambda>":
        raise ValueError("You must provide a name for lambda functions")

    func_doc = description or fn.__doc__ or ""
    is_async = is_async_callable(fn)

    if context_kwarg is None:  # pragma: no branch
        context_kwarg = find_context_parameter(fn)

    func_arg_metadata = func_metadata(
        fn,
        skip_names=[context_kwarg] if context_kwarg is not None else [],
        structured_output=structured_output,
    )
    parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)

    return cls(
        fn=fn,
        name=func_name,
        title=title,
        description=func_doc,
        parameters=parameters,
        fn_metadata=func_arg_metadata,
        is_async=is_async,
        context_kwarg=context_kwarg,
        annotations=annotations,
        icons=icons,
        meta=meta,
    )

run async 

run(
    arguments: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
    convert_result: bool = False,
) -> Any

Run the tool with arguments.

Raises:

Type Description
ToolError

If the tool function raises during execution.
Source code in src/mcp/server/mcpserver/tools/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
async def run(
    self,
    arguments: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
    convert_result: bool = False,
) -> Any:
    """Run the tool with arguments.

    Raises:
        ToolError: If the tool function raises during execution.
    """
    try:
        result = await self.fn_metadata.call_fn_with_arg_validation(
            self.fn,
            self.is_async,
            arguments,
            {self.context_kwarg: context} if self.context_kwarg is not None else None,
        )

        if convert_result:
            result = self.fn_metadata.convert_result(result)

        return result
    except UrlElicitationRequiredError:
        # Re-raise UrlElicitationRequiredError so it can be properly handled
        # as an MCP error response with code -32042
        raise
    except Exception as e:
        raise ToolError(f"Error executing tool {self.name}: {e}") from e

ToolManager

Manages MCPServer tools.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class ToolManager:
    """Manages MCPServer tools."""

    def __init__(self, warn_on_duplicate_tools: bool = True, *, tools: list[Tool] | None = None):
        self._tools: dict[str, Tool] = {}
        for tool in tools or ():
            if warn_on_duplicate_tools and tool.name in self._tools:
                logger.warning(f"Tool already exists: {tool.name}")
            self._tools[tool.name] = tool

        self.warn_on_duplicate_tools = warn_on_duplicate_tools

    def get_tool(self, name: str) -> Tool | None:
        """Get tool by name."""
        return self._tools.get(name)

    def list_tools(self) -> list[Tool]:
        """List all registered tools."""
        return list(self._tools.values())

    def add_tool(
        self,
        fn: Callable[..., Any],
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        annotations: ToolAnnotations | None = None,
        icons: list[Icon] | None = None,
        meta: dict[str, Any] | None = None,
        structured_output: bool | None = None,
    ) -> Tool:
        """Add a tool to the server."""
        tool = Tool.from_function(
            fn,
            name=name,
            title=title,
            description=description,
            annotations=annotations,
            icons=icons,
            meta=meta,
            structured_output=structured_output,
        )
        existing = self._tools.get(tool.name)
        if existing:
            if self.warn_on_duplicate_tools:
                logger.warning(f"Tool already exists: {tool.name}")
            return existing
        self._tools[tool.name] = tool
        return tool

    def remove_tool(self, name: str) -> None:
        """Remove a tool by name."""
        if name not in self._tools:
            raise ToolError(f"Unknown tool: {name}")
        del self._tools[name]

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any],
        context: Context[LifespanContextT, RequestT],
        convert_result: bool = False,
    ) -> Any:
        """Call a tool by name with arguments."""
        tool = self.get_tool(name)
        if not tool:
            raise ToolError(f"Unknown tool: {name}")

        return await tool.run(arguments, context, convert_result=convert_result)

get_tool 

get_tool(name: str) -> Tool | None

Get tool by name.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
30
31
32
def get_tool(self, name: str) -> Tool | None:
    """Get tool by name."""
    return self._tools.get(name)

list_tools 

list_tools() -> list[Tool]

List all registered tools.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
34
35
36
def list_tools(self) -> list[Tool]:
    """List all registered tools."""
    return list(self._tools.values())

add_tool 

add_tool(
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Tool

Add a tool to the server.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def add_tool(
    self,
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Tool:
    """Add a tool to the server."""
    tool = Tool.from_function(
        fn,
        name=name,
        title=title,
        description=description,
        annotations=annotations,
        icons=icons,
        meta=meta,
        structured_output=structured_output,
    )
    existing = self._tools.get(tool.name)
    if existing:
        if self.warn_on_duplicate_tools:
            logger.warning(f"Tool already exists: {tool.name}")
        return existing
    self._tools[tool.name] = tool
    return tool

remove_tool 

remove_tool(name: str) -> None

Remove a tool by name.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
68
69
70
71
72
def remove_tool(self, name: str) -> None:
    """Remove a tool by name."""
    if name not in self._tools:
        raise ToolError(f"Unknown tool: {name}")
    del self._tools[name]

call_tool async 

call_tool(
    name: str,
    arguments: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
    convert_result: bool = False,
) -> Any

Call a tool by name with arguments.

Source code in src/mcp/server/mcpserver/tools/tool_manager.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
    convert_result: bool = False,
) -> Any:
    """Call a tool by name with arguments."""
    tool = self.get_tool(name)
    if not tool:
        raise ToolError(f"Unknown tool: {name}")

    return await tool.run(arguments, context, convert_result=convert_result)