Skip to content

MCP Tools

MCP Tool.

MCPTool

Bases: AsyncBaseTool

MCP Tool Class.

Source code in src/llm_agents_from_scratch/tools/mcp/tool.py 
12
13
14
15
16
17
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
87
88
89
90
91
92
93
class MCPTool(AsyncBaseTool):
    """MCP Tool Class."""

    def __init__(
        self,
        provider: MCPToolProvider,
        name: str,
        desc: str,
        params_json_schema: dict[str, Any],
        additional_annotations: ToolAnnotations | None = None,
    ) -> None:
        """Initialize an MCP Tool.

        Note:
            It is highly recommended to use `MCPToolProvider.get_tools()` to
            create MCPTool instances. It automatically names tools as
            "mcp__{provider_name}__{server_tool_name}" to avoid collisions
            across providers. When the tool is invoked, the provider prefix
            is stripped to call the tool by its original server-side name.

        Args:
            provider (MCPToolProvider): The provider that owns this tool and
                manages the connection to the MCP server.
            name (str): The fully qualified tool name. When created via
                `provider.get_tools()`, this follows the format
                "mcp__{provider_name}__{server_tool_name}".
            desc (str): A description of what the tool does.
            params_json_schema (dict[str, Any]): JSON schema defining the
                tool's input parameters.
            additional_annotations (ToolAnnotations | None, optional):
                Additional MCP tool annotations (hints for clients).
                Defaults to None.
        """
        self.provider = provider
        self._name = name
        self._desc = desc
        self._params_json_schema = params_json_schema
        self.additional_annotations = additional_annotations

    @property
    def name(self) -> str:
        """Implements name property."""
        return self._name

    @property
    def description(self) -> str:
        """Implements description property."""
        return self._desc

    @property
    def parameters_json_schema(self) -> dict[str, Any]:
        """JSON schema for tool parameters."""
        return self._params_json_schema

    async def __call__(
        self,
        tool_call: ToolCall,
        *args: Any,
        **kwargs: Any,
    ) -> ToolCallResult:
        """Asynchronously execute the MCP tool call.

        Args:
            tool_call (ToolCall): The tool call to execute.
            *args (Any): Additional positional arguments forwarded to the tool.
            **kwargs (Any): Additional keyword arguments forwarded to the tool.

        Returns:
            ToolCallResult: The tool call result.
        """
        session = await self.provider.session()
        # call tool
        result = await session.call_tool(
            name=self.name.removeprefix(f"mcp__{self.provider.name}__"),
            arguments=tool_call.arguments,
        )

        return ToolCallResult(
            tool_call_id=tool_call.id_,
            content=[el.model_dump() for el in result.content],
            error=result.isError,
        )

name property 

name

Implements name property.

description property 

description

Implements description property.

parameters_json_schema property 

parameters_json_schema

JSON schema for tool parameters.

__init__ 

__init__(
    provider,
    name,
    desc,
    params_json_schema,
    additional_annotations=None,
)

Initialize an MCP Tool.

Note

It is highly recommended to use MCPToolProvider.get_tools() to create MCPTool instances. It automatically names tools as "mcp__{provider_name}__{server_tool_name}" to avoid collisions across providers. When the tool is invoked, the provider prefix is stripped to call the tool by its original server-side name.

Parameters:

Name Type Description Default
provider MCPToolProvider

The provider that owns this tool and manages the connection to the MCP server.

 required
name str

The fully qualified tool name. When created via provider.get_tools(), this follows the format "mcp__{provider_name}__{server_tool_name}".

 required
desc str

A description of what the tool does.

 required
params_json_schema dict[str, Any]

JSON schema defining the tool's input parameters.

 required
additional_annotations ToolAnnotations | None

Additional MCP tool annotations (hints for clients). Defaults to None.

 None
Source code in src/llm_agents_from_scratch/tools/mcp/tool.py 
15
16
17
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
def __init__(
    self,
    provider: MCPToolProvider,
    name: str,
    desc: str,
    params_json_schema: dict[str, Any],
    additional_annotations: ToolAnnotations | None = None,
) -> None:
    """Initialize an MCP Tool.

    Note:
        It is highly recommended to use `MCPToolProvider.get_tools()` to
        create MCPTool instances. It automatically names tools as
        "mcp__{provider_name}__{server_tool_name}" to avoid collisions
        across providers. When the tool is invoked, the provider prefix
        is stripped to call the tool by its original server-side name.

    Args:
        provider (MCPToolProvider): The provider that owns this tool and
            manages the connection to the MCP server.
        name (str): The fully qualified tool name. When created via
            `provider.get_tools()`, this follows the format
            "mcp__{provider_name}__{server_tool_name}".
        desc (str): A description of what the tool does.
        params_json_schema (dict[str, Any]): JSON schema defining the
            tool's input parameters.
        additional_annotations (ToolAnnotations | None, optional):
            Additional MCP tool annotations (hints for clients).
            Defaults to None.
    """
    self.provider = provider
    self._name = name
    self._desc = desc
    self._params_json_schema = params_json_schema
    self.additional_annotations = additional_annotations

__call__ async 

__call__(tool_call, *args, **kwargs)

Asynchronously execute the MCP tool call.

Parameters:

Name Type Description Default
tool_call ToolCall

The tool call to execute.

 required
*args Any

Additional positional arguments forwarded to the tool.

 ()
**kwargs Any

Additional keyword arguments forwarded to the tool.

 {}

Returns:

Name Type Description
ToolCallResult ToolCallResult

The tool call result.
Source code in src/llm_agents_from_scratch/tools/mcp/tool.py 
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
async def __call__(
    self,
    tool_call: ToolCall,
    *args: Any,
    **kwargs: Any,
) -> ToolCallResult:
    """Asynchronously execute the MCP tool call.

    Args:
        tool_call (ToolCall): The tool call to execute.
        *args (Any): Additional positional arguments forwarded to the tool.
        **kwargs (Any): Additional keyword arguments forwarded to the tool.

    Returns:
        ToolCallResult: The tool call result.
    """
    session = await self.provider.session()
    # call tool
    result = await session.call_tool(
        name=self.name.removeprefix(f"mcp__{self.provider.name}__"),
        arguments=tool_call.arguments,
    )

    return ToolCallResult(
        tool_call_id=tool_call.id_,
        content=[el.model_dump() for el in result.content],
        error=result.isError,
    )

MCP Tool Provider.

MCPToolProvider

MCP Tool Provider class.

Source code in src/llm_agents_from_scratch/tools/mcp/provider.py 
 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
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class MCPToolProvider:
    """MCP Tool Provider class."""

    def __init__(
        self,
        name: str,
        stdio_params: StdioServerParameters | None = None,
        streamable_http_url: str | None = None,
        streamable_http_headers: dict[str, str] | None = None,
    ) -> None:
        """Initialize an MCPToolProvider.

        Args:
            name (str): A name identifier for this provider. Used to prefix
                tool names when creating MCPTool instances (e.g.,
                "mcp__{name}__{tool_name}").
            stdio_params (StdioServerParameters | None, optional): Parameters
                for connecting to an MCP server via stdio. If both this and
                `streamable_http_url` are provided, stdio will be used and
                HTTP will be ignored. Defaults to None.
            streamable_http_url (str | None, optional): URL for connecting to
                an MCP server via HTTP. Only used if `stdio_params` is None.
                Defaults to None.
            streamable_http_headers (dict[str, str] | None, optional): HTTP
                headers to include with every request to the MCP server (e.g.,
                ``{"Authorization": "Bearer <token>"}``). Only used when
                connecting via HTTP. Defaults to None.

        Raises:
            MissingMCPServerParamsError: If neither `stdio_params` nor
                `streamable_http_url` is provided.

        Warns:
            MCPWarning: Emitted if both `stdio_params` and
                `streamable_http_url` are provided (stdio will be prioritized).
        """
        if (stdio_params is None) and (streamable_http_url is None):
            msg = (
                "You must supply at least one of `stdio_params` or "
                "`streamable_http_url` to connect to an MCP server."
            )
            raise MissingMCPServerParamsError(msg)

        if stdio_params and streamable_http_url:
            msg = (
                "Both `stdio_params` and `streamable_http_url` were provided; "
                "`stdio_params` will be used and `streamable_http_url` ignored."
            )
            warnings.warn(msg, MCPWarning, stacklevel=2)

        self.name = name
        self.stdio_params = stdio_params
        self.streamable_http_url = streamable_http_url
        self.streamable_http_headers = streamable_http_headers
        # initialize session management attributes
        self._shutdown_event = asyncio.Event()
        self._session_ready = asyncio.Event()
        self._session_task: asyncio.Task | None = None
        self._session: ClientSession | None = None

    async def _create_session(self) -> None:
        """Create and maintain persistent session until shutdown signal.

        This method runs as a background task, keeping the MCP session
        alive by holding the context managers open. When the shutdown
        event is set, the context managers exit gracefully.
        """
        if self.stdio_params:
            async with stdio_client(self.stdio_params) as (read, write):  # noqa: SIM117
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    self._session = session
                    self._session_ready.set()

                    # Wait for shutdown signal
                    await self._shutdown_event.wait()
        else:
            async with streamablehttp_client(  # noqa: SIM117
                self.streamable_http_url,
                self.streamable_http_headers,
            ) as (
                read_stream,
                write_stream,
                _,
            ):
                async with ClientSession(read_stream, write_stream) as session:
                    await session.initialize()
                    self._session = session
                    self._session_ready.set()

                    # Wait for shutdown signal
                    await self._shutdown_event.wait()

    async def session(self) -> ClientSession:
        """Get the persistent session.

        Returns:
            ClientSession: An initialized MCP client session.

        Raises:
            Exception: Re-raises any exception encountered during session
                creation (e.g. invalid server path, connection refused).

        Note:
            This method uses lazy initialization - the session is created
            on the first call and reused for subsequent calls.
        """
        if not self._session_ready.is_set():
            self._session_task = asyncio.create_task(self._create_session())
            session_ready_task = asyncio.create_task(self._session_ready.wait())

            # Wait for the first of the these two tasks to complete.
            # If session_ready_task completes first, then everything is good and
            # the session created successfully. Otherwise, the session_ready
            # event was never set, meaning an error was encountered.
            done, _pending = await asyncio.wait(
                [self._session_task, session_ready_task],
                return_when=asyncio.FIRST_COMPLETED,
            )

            if self._session_task in done:
                session_ready_task.cancel()  # clean-up session ready task
                self._session_task.result()  # re-raises the encountered error
        return self._session  # type: ignore[return-value]

    async def get_tools(self) -> list["MCPTool"]:
        """Fetch tools from the MCP server and create MCPTool instances.

        Returns:
            list[MCPTool]: A list of MCPTool instances representing the
                tools available from the MCP server.
        """
        from llm_agents_from_scratch.tools.mcp.tool import (  # noqa: PLC0415
            MCPTool,
        )

        session = await self.session()
        response = await session.list_tools()

        return [
            MCPTool(
                provider=self,
                name=f"mcp__{self.name}__{tool.name}",
                desc=tool.description,
                params_json_schema=tool.inputSchema,
                additional_annotations=tool.annotations,
            )
            for tool in response.tools
        ]

    async def close(self) -> None:
        """Close the persistent session and clean up resources.

        Note:
            For short-lived scripts, calling close() is optional as the OS
            will clean up subprocess resources when your program exits.
            For long-running applications, you should call close() to
            prevent resource leaks.
        """
        if self._session_task:
            self._shutdown_event.set()
            try:
                await self._session_task
            finally:
                self._session = None
                self._session_task = None
                self._shutdown_event.clear()
                self._session_ready.clear()

__init__ 

__init__(
    name,
    stdio_params=None,
    streamable_http_url=None,
    streamable_http_headers=None,
)

Initialize an MCPToolProvider.

Parameters:

Name Type Description Default
name str

A name identifier for this provider. Used to prefix tool names when creating MCPTool instances (e.g., "mcp__{name}__{tool_name}").

 required
stdio_params StdioServerParameters | None

Parameters for connecting to an MCP server via stdio. If both this and streamable_http_url are provided, stdio will be used and HTTP will be ignored. Defaults to None.

 None
streamable_http_url str | None

URL for connecting to an MCP server via HTTP. Only used if stdio_params is None. Defaults to None.

 None
streamable_http_headers dict[str, str] | None

HTTP headers to include with every request to the MCP server (e.g., {"Authorization": "Bearer <token>"}). Only used when connecting via HTTP. Defaults to None.

 None

Raises:

Type Description
MissingMCPServerParamsError

If neither stdio_params nor streamable_http_url is provided.

Warns:

Type Description
MCPWarning

Emitted if both stdio_params and streamable_http_url are provided (stdio will be prioritized).
Source code in src/llm_agents_from_scratch/tools/mcp/provider.py 
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
def __init__(
    self,
    name: str,
    stdio_params: StdioServerParameters | None = None,
    streamable_http_url: str | None = None,
    streamable_http_headers: dict[str, str] | None = None,
) -> None:
    """Initialize an MCPToolProvider.

    Args:
        name (str): A name identifier for this provider. Used to prefix
            tool names when creating MCPTool instances (e.g.,
            "mcp__{name}__{tool_name}").
        stdio_params (StdioServerParameters | None, optional): Parameters
            for connecting to an MCP server via stdio. If both this and
            `streamable_http_url` are provided, stdio will be used and
            HTTP will be ignored. Defaults to None.
        streamable_http_url (str | None, optional): URL for connecting to
            an MCP server via HTTP. Only used if `stdio_params` is None.
            Defaults to None.
        streamable_http_headers (dict[str, str] | None, optional): HTTP
            headers to include with every request to the MCP server (e.g.,
            ``{"Authorization": "Bearer <token>"}``). Only used when
            connecting via HTTP. Defaults to None.

    Raises:
        MissingMCPServerParamsError: If neither `stdio_params` nor
            `streamable_http_url` is provided.

    Warns:
        MCPWarning: Emitted if both `stdio_params` and
            `streamable_http_url` are provided (stdio will be prioritized).
    """
    if (stdio_params is None) and (streamable_http_url is None):
        msg = (
            "You must supply at least one of `stdio_params` or "
            "`streamable_http_url` to connect to an MCP server."
        )
        raise MissingMCPServerParamsError(msg)

    if stdio_params and streamable_http_url:
        msg = (
            "Both `stdio_params` and `streamable_http_url` were provided; "
            "`stdio_params` will be used and `streamable_http_url` ignored."
        )
        warnings.warn(msg, MCPWarning, stacklevel=2)

    self.name = name
    self.stdio_params = stdio_params
    self.streamable_http_url = streamable_http_url
    self.streamable_http_headers = streamable_http_headers
    # initialize session management attributes
    self._shutdown_event = asyncio.Event()
    self._session_ready = asyncio.Event()
    self._session_task: asyncio.Task | None = None
    self._session: ClientSession | None = None

session async 

session()

Get the persistent session.

Returns:

Name Type Description
ClientSession ClientSession

An initialized MCP client session.

Raises:

Type Description
Exception

Re-raises any exception encountered during session creation (e.g. invalid server path, connection refused).
Note

This method uses lazy initialization - the session is created on the first call and reused for subsequent calls.

Source code in src/llm_agents_from_scratch/tools/mcp/provider.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def session(self) -> ClientSession:
    """Get the persistent session.

    Returns:
        ClientSession: An initialized MCP client session.

    Raises:
        Exception: Re-raises any exception encountered during session
            creation (e.g. invalid server path, connection refused).

    Note:
        This method uses lazy initialization - the session is created
        on the first call and reused for subsequent calls.
    """
    if not self._session_ready.is_set():
        self._session_task = asyncio.create_task(self._create_session())
        session_ready_task = asyncio.create_task(self._session_ready.wait())

        # Wait for the first of the these two tasks to complete.
        # If session_ready_task completes first, then everything is good and
        # the session created successfully. Otherwise, the session_ready
        # event was never set, meaning an error was encountered.
        done, _pending = await asyncio.wait(
            [self._session_task, session_ready_task],
            return_when=asyncio.FIRST_COMPLETED,
        )

        if self._session_task in done:
            session_ready_task.cancel()  # clean-up session ready task
            self._session_task.result()  # re-raises the encountered error
    return self._session  # type: ignore[return-value]

get_tools async 

get_tools()

Fetch tools from the MCP server and create MCPTool instances.

Returns:

Type Description
list[MCPTool]

list[MCPTool]: A list of MCPTool instances representing the tools available from the MCP server.
Source code in src/llm_agents_from_scratch/tools/mcp/provider.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def get_tools(self) -> list["MCPTool"]:
    """Fetch tools from the MCP server and create MCPTool instances.

    Returns:
        list[MCPTool]: A list of MCPTool instances representing the
            tools available from the MCP server.
    """
    from llm_agents_from_scratch.tools.mcp.tool import (  # noqa: PLC0415
        MCPTool,
    )

    session = await self.session()
    response = await session.list_tools()

    return [
        MCPTool(
            provider=self,
            name=f"mcp__{self.name}__{tool.name}",
            desc=tool.description,
            params_json_schema=tool.inputSchema,
            additional_annotations=tool.annotations,
        )
        for tool in response.tools
    ]

close async 

close()

Close the persistent session and clean up resources.

Note

For short-lived scripts, calling close() is optional as the OS will clean up subprocess resources when your program exits. For long-running applications, you should call close() to prevent resource leaks.

Source code in src/llm_agents_from_scratch/tools/mcp/provider.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def close(self) -> None:
    """Close the persistent session and clean up resources.

    Note:
        For short-lived scripts, calling close() is optional as the OS
        will clean up subprocess resources when your program exits.
        For long-running applications, you should call close() to
        prevent resource leaks.
    """
    if self._session_task:
        self._shutdown_event.set()
        try:
            await self._session_task
        finally:
            self._session = None
            self._session_task = None
            self._shutdown_event.clear()
            self._session_ready.clear()