Skip to content

Core

Core module for the MCPAdapt library.

This module contains the core functionality for the MCPAdapt library. It provides the basic interfaces and classes for adapting tools from MCP to the desired Agent framework.

ToolAdapter

Bases: ABC

A basic interface for adapting tools from MCP to the desired Agent framework.

Source code in src/mcpadapt/core.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
class ToolAdapter(ABC):
    """A basic interface for adapting tools from MCP to the desired Agent framework."""

    @abstractmethod
    def adapt(
        self,
        func: Callable[[dict | None], mcp.types.CallToolResult],
        mcp_tool: mcp.types.Tool,
    ) -> Any:
        """Adapt a single tool from MCP to the desired Agent framework.

        The MCP protocol will provide a name, description and inputSchema in JSON Schema
        format. This needs to be adapted to the desired Agent framework.

        Note that the function is synchronous (not a coroutine) you can use
        :meth:`ToolAdapter.async_adapt` if you need to use the tool asynchronously.

        Args:
            func: The function to be called (will call the tool via the MCP protocol).
            mcp_tool: The tool to adapt.

        Returns:
            The adapted tool in the agentic framework of choice.
        """
        pass

    def async_adapt(
        self,
        afunc: Callable[[dict | None], Coroutine[Any, Any, mcp.types.CallToolResult]],
        mcp_tool: mcp.types.Tool,
    ) -> Any:
        """Adapt a single tool from MCP to the desired Agent framework.

        The MCP protocol will provide a name, description and inputSchema in JSON Schema
        format. This needs to be adapted to the desired Agent framework.

        Note that the function is asynchronous (a coroutine) you can use
        :meth:`ToolAdapter.adapt` if you need to use the tool synchronously.

        Args:
            afunc: The coroutine to be called.
            mcp_tool: The tool to adapt.

        Returns:
            The adapted tool in the agentic framework of choice.
        """
        raise NotImplementedError(
            "Async adaptation is not supported for this Agent framework."
        )

adapt abstractmethod 

adapt(
    func: Callable[[dict | None], CallToolResult],
    mcp_tool: Tool,
) -> Any

Adapt a single tool from MCP to the desired Agent framework.

The MCP protocol will provide a name, description and inputSchema in JSON Schema format. This needs to be adapted to the desired Agent framework.

Note that the function is synchronous (not a coroutine) you can use :meth:ToolAdapter.async_adapt if you need to use the tool asynchronously.

Parameters:

Name Type Description Default
func Callable[[dict | None], CallToolResult]

The function to be called (will call the tool via the MCP protocol).

 required
mcp_tool Tool

The tool to adapt.

 required

Returns:

Type Description
Any

The adapted tool in the agentic framework of choice.
Source code in src/mcpadapt/core.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@abstractmethod
def adapt(
    self,
    func: Callable[[dict | None], mcp.types.CallToolResult],
    mcp_tool: mcp.types.Tool,
) -> Any:
    """Adapt a single tool from MCP to the desired Agent framework.

    The MCP protocol will provide a name, description and inputSchema in JSON Schema
    format. This needs to be adapted to the desired Agent framework.

    Note that the function is synchronous (not a coroutine) you can use
    :meth:`ToolAdapter.async_adapt` if you need to use the tool asynchronously.

    Args:
        func: The function to be called (will call the tool via the MCP protocol).
        mcp_tool: The tool to adapt.

    Returns:
        The adapted tool in the agentic framework of choice.
    """
    pass

async_adapt 

async_adapt(
    afunc: Callable[
        [dict | None], Coroutine[Any, Any, CallToolResult]
    ],
    mcp_tool: Tool,
) -> Any

Adapt a single tool from MCP to the desired Agent framework.

The MCP protocol will provide a name, description and inputSchema in JSON Schema format. This needs to be adapted to the desired Agent framework.

Note that the function is asynchronous (a coroutine) you can use :meth:ToolAdapter.adapt if you need to use the tool synchronously.

Parameters:

Name Type Description Default
afunc Callable[[dict | None], Coroutine[Any, Any, CallToolResult]]

The coroutine to be called.

 required
mcp_tool Tool

The tool to adapt.

 required

Returns:

Type Description
Any

The adapted tool in the agentic framework of choice.
Source code in src/mcpadapt/core.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def async_adapt(
    self,
    afunc: Callable[[dict | None], Coroutine[Any, Any, mcp.types.CallToolResult]],
    mcp_tool: mcp.types.Tool,
) -> Any:
    """Adapt a single tool from MCP to the desired Agent framework.

    The MCP protocol will provide a name, description and inputSchema in JSON Schema
    format. This needs to be adapted to the desired Agent framework.

    Note that the function is asynchronous (a coroutine) you can use
    :meth:`ToolAdapter.adapt` if you need to use the tool synchronously.

    Args:
        afunc: The coroutine to be called.
        mcp_tool: The tool to adapt.

    Returns:
        The adapted tool in the agentic framework of choice.
    """
    raise NotImplementedError(
        "Async adaptation is not supported for this Agent framework."
    )

MCPAdapt

The main class for adapting MCP tools to the desired Agent framework.

This class can be used either as a sync or async context manager.

If running synchronously, it will run the MCP server in a separate thread and take care of making the tools synchronous without blocking the server.

If running asynchronously, it will use the async context manager and return async tools.

Dependening on what your Agent framework supports choose the approriate method. If async is supported it is recommended.

Important Note: adapters need to implement the async_adapt method to support async tools.

Usage:

sync usage

with MCPAdapt(StdioServerParameters(command="uv", args=["run", "src/echo.py"]), SmolAgentAdapter()) as tools: print(tools)

async usage

async with MCPAdapt(StdioServerParameters(command="uv", args=["run", "src/echo.py"]), SmolAgentAdapter()) as tools: print(tools)

async usage with sse

async with MCPAdapt({"host": "127.0.0.1", "port": 8000}, SmolAgentAdapter()) as tools: print(tools)

Source code in src/mcpadapt/core.py 
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class MCPAdapt:
    """The main class for adapting MCP tools to the desired Agent framework.

    This class can be used either as a sync or async context manager.

    If running synchronously, it will run the MCP server in a separate thread and take
    care of making the tools synchronous without blocking the server.

    If running asynchronously, it will use the async context manager and return async
    tools.

    Dependening on what your Agent framework supports choose the approriate method. If
    async is supported it is recommended.

    Important Note: adapters need to implement the async_adapt method to support async
    tools.

    Usage:
    >>> # sync usage
    >>> with MCPAdapt(StdioServerParameters(command="uv", args=["run", "src/echo.py"]), SmolAgentAdapter()) as tools:
    >>>     print(tools)

    >>> # async usage
    >>> async with MCPAdapt(StdioServerParameters(command="uv", args=["run", "src/echo.py"]), SmolAgentAdapter()) as tools:
    >>>     print(tools)

    >>> # async usage with sse
    >>> async with MCPAdapt({"host": "127.0.0.1", "port": 8000}, SmolAgentAdapter()) as tools:
    >>>     print(tools)
    """

    def __init__(
        self,
        serverparams: StdioServerParameters | dict[str, Any],
        adapter: ToolAdapter,
        connect_timeout: int = 30,
    ):
        """
        Manage the MCP server / client lifecycle and expose tools adapted with the adapter.

        Args:
            serverparams (StdioServerParameters | dict[str, Any]): MCP server parameters (stdio or sse).
            adapter (ToolAdapter): Adapter to use to convert MCP tools call into agentic framework tools.
            connect_timeout (int): Connection timeout in seconds to the mcp server (default is 30s).

        Raises:
            TimeoutError: When the connection to the mcp server time out.
        """
        # attributes we receive from the user.
        self.serverparams = serverparams
        self.adapter = adapter

        # session and tools get set by the async loop during initialization.
        self.session: ClientSession | None = None
        self.mcp_tools: list[mcp.types.Tool] | None = None

        # all attributes used to manage the async loop and separate thread.
        self.loop = asyncio.new_event_loop()
        self.task = None
        self.ready = threading.Event()
        self.thread = threading.Thread(target=self._run_loop, daemon=True)

        # start the loop in a separate thread and wait till ready synchronously.
        self.thread.start()
        # check connection to mcp server is ready
        if not self.ready.wait(timeout=connect_timeout):
            raise TimeoutError(
                f"Couldn't connect to the MCP server after {connect_timeout} seconds"
            )

    def _run_loop(self):
        """Runs the event loop in a separate thread (for synchronous usage)."""
        asyncio.set_event_loop(self.loop)

        async def setup():
            async with mcptools(self.serverparams) as (session, tools):
                self.session, self.mcp_tools = session, tools
                self.ready.set()  # Signal initialization is complete
                await asyncio.Event().wait()  # Keep session alive until stopped

        self.task = self.loop.create_task(setup())
        try:
            self.loop.run_until_complete(self.task)
        except asyncio.CancelledError:
            pass

    def tools(self) -> list[Any]:
        """Returns the tools from the MCP server adapted to the desired Agent framework.

        This is what is yielded if used as a context manager otherwise you can access it
        directly via this method.

        An equivalent async method is available if your Agent framework supports it:
        see :meth:`atools`.

        """
        if not self.session:
            raise RuntimeError("Session not initialized")

        def _sync_call_tool(
            name: str, arguments: dict | None = None
        ) -> mcp.types.CallToolResult:
            return asyncio.run_coroutine_threadsafe(
                self.session.call_tool(name, arguments), self.loop
            ).result()

        return [
            self.adapter.adapt(partial(_sync_call_tool, tool.name), tool)
            for tool in self.mcp_tools
        ]

    def close(self):
        """Clean up resources and stop the client."""
        if self.task and not self.task.done():
            self.loop.call_soon_threadsafe(self.task.cancel)
        self.thread.join()  # will wait until the task is cancelled to join thread (as it's blocked Event().wait())
        self.loop.close()  # we won't be using the loop anymore we can safely close it

    def __enter__(self):
        return self.tools()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    # -- add support for async context manager as well if the agent framework supports it.
    def atools(self) -> list[Any]:
        """Returns the tools from the MCP server adapted to the desired Agent framework.

        This is what is yielded if used as an async context manager otherwise you can
        access it directly via this method.

        An equivalent async method is available if your Agent framework supports it:
        see :meth:`atools`.
        """
        return [
            self.adapter.async_adapt(partial(self.session.call_tool, tool.name), tool)
            for tool in self.mcp_tools
        ]

    async def __aenter__(self) -> list[Any]:
        self._ctxmanager = mcptools(self.serverparams)
        self.session, self.tools = await self._ctxmanager.__aenter__()
        return self.atools()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._ctxmanager.__aexit__(exc_type, exc_val, exc_tb)

__init__ 

__init__(
    serverparams: StdioServerParameters | dict[str, Any],
    adapter: ToolAdapter,
    connect_timeout: int = 30,
)

Manage the MCP server / client lifecycle and expose tools adapted with the adapter.

Parameters:

Name Type Description Default
serverparams StdioServerParameters | dict[str, Any]

MCP server parameters (stdio or sse).

 required
adapter ToolAdapter

Adapter to use to convert MCP tools call into agentic framework tools.

 required
connect_timeout int

Connection timeout in seconds to the mcp server (default is 30s).

 30

Raises:

Type Description
TimeoutError

When the connection to the mcp server time out.
Source code in src/mcpadapt/core.py 
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
def __init__(
    self,
    serverparams: StdioServerParameters | dict[str, Any],
    adapter: ToolAdapter,
    connect_timeout: int = 30,
):
    """
    Manage the MCP server / client lifecycle and expose tools adapted with the adapter.

    Args:
        serverparams (StdioServerParameters | dict[str, Any]): MCP server parameters (stdio or sse).
        adapter (ToolAdapter): Adapter to use to convert MCP tools call into agentic framework tools.
        connect_timeout (int): Connection timeout in seconds to the mcp server (default is 30s).

    Raises:
        TimeoutError: When the connection to the mcp server time out.
    """
    # attributes we receive from the user.
    self.serverparams = serverparams
    self.adapter = adapter

    # session and tools get set by the async loop during initialization.
    self.session: ClientSession | None = None
    self.mcp_tools: list[mcp.types.Tool] | None = None

    # all attributes used to manage the async loop and separate thread.
    self.loop = asyncio.new_event_loop()
    self.task = None
    self.ready = threading.Event()
    self.thread = threading.Thread(target=self._run_loop, daemon=True)

    # start the loop in a separate thread and wait till ready synchronously.
    self.thread.start()
    # check connection to mcp server is ready
    if not self.ready.wait(timeout=connect_timeout):
        raise TimeoutError(
            f"Couldn't connect to the MCP server after {connect_timeout} seconds"
        )

tools 

tools() -> list[Any]

Returns the tools from the MCP server adapted to the desired Agent framework.

This is what is yielded if used as a context manager otherwise you can access it directly via this method.

An equivalent async method is available if your Agent framework supports it: see :meth:atools.

Source code in src/mcpadapt/core.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def tools(self) -> list[Any]:
    """Returns the tools from the MCP server adapted to the desired Agent framework.

    This is what is yielded if used as a context manager otherwise you can access it
    directly via this method.

    An equivalent async method is available if your Agent framework supports it:
    see :meth:`atools`.

    """
    if not self.session:
        raise RuntimeError("Session not initialized")

    def _sync_call_tool(
        name: str, arguments: dict | None = None
    ) -> mcp.types.CallToolResult:
        return asyncio.run_coroutine_threadsafe(
            self.session.call_tool(name, arguments), self.loop
        ).result()

    return [
        self.adapter.adapt(partial(_sync_call_tool, tool.name), tool)
        for tool in self.mcp_tools
    ]

close 

close()

Clean up resources and stop the client.

Source code in src/mcpadapt/core.py 
220
221
222
223
224
225
def close(self):
    """Clean up resources and stop the client."""
    if self.task and not self.task.done():
        self.loop.call_soon_threadsafe(self.task.cancel)
    self.thread.join()  # will wait until the task is cancelled to join thread (as it's blocked Event().wait())
    self.loop.close()  # we won't be using the loop anymore we can safely close it

atools 

atools() -> list[Any]

Returns the tools from the MCP server adapted to the desired Agent framework.

This is what is yielded if used as an async context manager otherwise you can access it directly via this method.

An equivalent async method is available if your Agent framework supports it: see :meth:atools.

Source code in src/mcpadapt/core.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
def atools(self) -> list[Any]:
    """Returns the tools from the MCP server adapted to the desired Agent framework.

    This is what is yielded if used as an async context manager otherwise you can
    access it directly via this method.

    An equivalent async method is available if your Agent framework supports it:
    see :meth:`atools`.
    """
    return [
        self.adapter.async_adapt(partial(self.session.call_tool, tool.name), tool)
        for tool in self.mcp_tools
    ]

mcptools async 

mcptools(
    serverparams: StdioServerParameters | dict[str, Any],
) -> tuple[ClientSession, list[Tool]]

Async context manager that yields tools from an MCP server.

Note: the session can be then used to call tools on the MCP server but it's async. Use MCPAdapt instead if you need to use the tools synchronously.

Parameters:

Name Type Description Default
serverparams StdioServerParameters | dict[str, Any]

Parameters passed to either the stdio client or sse client. * if StdioServerParameters, run the MCP server using the stdio protocol. * if dict, assume the dict corresponds to parameters to an sse MCP server.

 required

Yields:

Type Description
tuple[ClientSession, list[Tool]]

A list of tools available on the MCP server.

Usage:

async with mcptools(StdioServerParameters(command="uv", args=["run", "src/echo.py"])) as (session, tools): print(tools)

Source code in src/mcpadapt/core.py 
 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
@asynccontextmanager
async def mcptools(
    serverparams: StdioServerParameters | dict[str, Any],
) -> tuple[ClientSession, list[mcp.types.Tool]]:
    """Async context manager that yields tools from an MCP server.

    Note: the session can be then used to call tools on the MCP server but it's async.
    Use MCPAdapt instead if you need to use the tools synchronously.

    Args:
        serverparams: Parameters passed to either the stdio client or sse client.
            * if StdioServerParameters, run the MCP server using the stdio protocol.
            * if dict, assume the dict corresponds to parameters to an sse MCP server.

    Yields:
        A list of tools available on the MCP server.

    Usage:
    >>> async with mcptools(StdioServerParameters(command="uv", args=["run", "src/echo.py"])) as (session, tools):
    >>>     print(tools)
    """
    if isinstance(serverparams, StdioServerParameters):
        client = stdio_client(serverparams)
    elif isinstance(serverparams, dict):
        client = sse_client(**serverparams)
    else:
        raise ValueError(
            f"Invalid serverparams, expected StdioServerParameters or dict found `{type(serverparams)}`"
        )

    async with client as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection and get the tools from the mcp server
            await session.initialize()
            tools = await session.list_tools()
            yield session, tools.tools