Skip to content

tool_manager

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)