Skip to content

Index

MCP Client module.

Transport

Bases: AbstractAsyncContextManager[TransportStreams], Protocol

Protocol for MCP transports.

A transport is an async context manager that yields read and write streams for bidirectional communication with an MCP server.

Source code in src/mcp/client/_transport.py 
16
17
18
19
20
21
class Transport(AbstractAsyncContextManager[TransportStreams], Protocol):
    """Protocol for MCP transports.

    A transport is an async context manager that yields read and write streams
    for bidirectional communication with an MCP server.
    """

Client dataclass

A high-level MCP client for connecting to MCP servers.

Supports in-memory transport for testing (pass a Server or MCPServer instance), Streamable HTTP transport (pass a URL string), or a custom Transport instance.

Example 
from mcp.client import Client
from mcp.server.mcpserver import MCPServer

server = MCPServer("test")

@server.tool()
def add(a: int, b: int) -> int:
    return a + b

async def main():
    async with Client(server) as client:
        result = await client.call_tool("add", {"a": 1, "b": 2})

asyncio.run(main())
Source code in src/mcp/client/client.py 
 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
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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@dataclass
class Client:
    """A high-level MCP client for connecting to MCP servers.

    Supports in-memory transport for testing (pass a Server or MCPServer instance),
    Streamable HTTP transport (pass a URL string), or a custom Transport instance.

    Example:
        ```python
        from mcp.client import Client
        from mcp.server.mcpserver import MCPServer

        server = MCPServer("test")

        @server.tool()
        def add(a: int, b: int) -> int:
            return a + b

        async def main():
            async with Client(server) as client:
                result = await client.call_tool("add", {"a": 1, "b": 2})

        asyncio.run(main())
        ```
    """

    server: Server[Any] | MCPServer | Transport | str
    """The MCP server to connect to.

    If the server is a `Server` or `MCPServer` instance, it will be wrapped in an `InMemoryTransport`.
    If the server is a URL string, it will be used as the URL for a `streamable_http_client` transport.
    If the server is a `Transport` instance, it will be used directly.
    """

    _: KW_ONLY

    # TODO(Marcelo): When do `raise_exceptions=True` actually raises?
    raise_exceptions: bool = False
    """Whether to raise exceptions from the server."""

    read_timeout_seconds: float | None = None
    """Timeout for read operations."""

    sampling_callback: SamplingFnT | None = None
    """Callback for handling sampling requests."""

    list_roots_callback: ListRootsFnT | None = None
    """Callback for handling list roots requests."""

    logging_callback: LoggingFnT | None = None
    """Callback for handling logging notifications."""

    # TODO(Marcelo): Why do we have both "callback" and "handler"?
    message_handler: MessageHandlerFnT | None = None
    """Callback for handling raw messages."""

    client_info: Implementation | None = None
    """Client implementation info to send to server."""

    elicitation_callback: ElicitationFnT | None = None
    """Callback for handling elicitation requests."""

    _session: ClientSession | None = field(init=False, default=None)
    _exit_stack: AsyncExitStack | None = field(init=False, default=None)
    _transport: Transport = field(init=False)

    def __post_init__(self) -> None:
        if isinstance(self.server, Server | MCPServer):
            self._transport = InMemoryTransport(self.server, raise_exceptions=self.raise_exceptions)
        elif isinstance(self.server, str):
            self._transport = streamable_http_client(self.server)
        else:
            self._transport = self.server

    async def __aenter__(self) -> Client:
        """Enter the async context manager."""
        if self._session is not None:
            raise RuntimeError("Client is already entered; cannot reenter")

        async with AsyncExitStack() as exit_stack:
            read_stream, write_stream = await exit_stack.enter_async_context(self._transport)

            self._session = await exit_stack.enter_async_context(
                ClientSession(
                    read_stream=read_stream,
                    write_stream=write_stream,
                    read_timeout_seconds=self.read_timeout_seconds,
                    sampling_callback=self.sampling_callback,
                    list_roots_callback=self.list_roots_callback,
                    logging_callback=self.logging_callback,
                    message_handler=self.message_handler,
                    client_info=self.client_info,
                    elicitation_callback=self.elicitation_callback,
                )
            )

            await self._session.initialize()

            # Transfer ownership to self for __aexit__ to handle
            self._exit_stack = exit_stack.pop_all()
            return self

    async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
        """Exit the async context manager."""
        if self._exit_stack:  # pragma: no branch
            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
        self._session = None

    @property
    def session(self) -> ClientSession:
        """Get the underlying ClientSession.

        This provides access to the full ClientSession API for advanced use cases.

        Raises:
            RuntimeError: If accessed before entering the context manager.
        """
        if self._session is None:
            raise RuntimeError("Client must be used within an async context manager")
        return self._session

    @property
    def initialize_result(self) -> InitializeResult:
        """The server's InitializeResult.

        Contains server_info, capabilities, instructions, and the negotiated protocol_version.
        Raises RuntimeError if accessed outside the context manager.
        """
        result = self.session.initialize_result
        if result is None:  # pragma: no cover
            raise RuntimeError("Client must be used within an async context manager")
        return result

    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Send a ping request to the server."""
        return await self.session.send_ping(meta=meta)

    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
    ) -> None:
        """Send a progress notification to the server."""
        await self.session.send_progress_notification(
            progress_token=progress_token,
            progress=progress,
            total=total,
            message=message,
        )

    async def set_logging_level(self, level: LoggingLevel, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Set the logging level on the server."""
        return await self.session.set_logging_level(level=level, meta=meta)

    async def list_resources(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> ListResourcesResult:
        """List available resources from the server."""
        return await self.session.list_resources(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

    async def list_resource_templates(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> ListResourceTemplatesResult:
        """List available resource templates from the server."""
        return await self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> ReadResourceResult:
        """Read a resource from the server.

        Args:
            uri: The URI of the resource to read.
            meta: Additional metadata for the request.

        Returns:
            The resource content.
        """
        return await self.session.read_resource(uri, meta=meta)

    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Subscribe to resource updates."""
        return await self.session.subscribe_resource(uri, meta=meta)

    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Unsubscribe from resource updates."""
        return await self.session.unsubscribe_resource(uri, meta=meta)

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> CallToolResult:
        """Call a tool on the server.

        Args:
            name: The name of the tool to call
            arguments: Arguments to pass to the tool
            read_timeout_seconds: Timeout for the tool call
            progress_callback: Callback for progress updates
            meta: Additional metadata for the request

        Returns:
            The tool result.
        """
        return await self.session.call_tool(
            name=name,
            arguments=arguments,
            read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
            meta=meta,
        )

    async def list_prompts(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> ListPromptsResult:
        """List available prompts from the server."""
        return await self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

    async def get_prompt(
        self, name: str, arguments: dict[str, str] | None = None, *, meta: RequestParamsMeta | None = None
    ) -> GetPromptResult:
        """Get a prompt from the server.

        Args:
            name: The name of the prompt
            arguments: Arguments to pass to the prompt
            meta: Additional metadata for the request

        Returns:
            The prompt content.
        """
        return await self.session.get_prompt(name=name, arguments=arguments, meta=meta)

    async def complete(
        self,
        ref: ResourceTemplateReference | PromptReference,
        argument: dict[str, str],
        context_arguments: dict[str, str] | None = None,
    ) -> CompleteResult:
        """Get completions for a prompt or resource template argument.

        Args:
            ref: Reference to the prompt or resource template
            argument: The argument to complete
            context_arguments: Additional context arguments

        Returns:
            Completion suggestions.
        """
        return await self.session.complete(ref=ref, argument=argument, context_arguments=context_arguments)

    async def list_tools(self, *, cursor: str | None = None, meta: RequestParamsMeta | None = None) -> ListToolsResult:
        """List available tools from the server."""
        return await self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

    async def send_roots_list_changed(self) -> None:
        """Send a notification that the roots list has changed."""
        # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
        await self.session.send_roots_list_changed()  # pragma: no cover

server instance-attribute 

server: Server[Any] | MCPServer | Transport | str

The MCP server to connect to.

If the server is a Server or MCPServer instance, it will be wrapped in an InMemoryTransport. If the server is a URL string, it will be used as the URL for a streamable_http_client transport. If the server is a Transport instance, it will be used directly.

raise_exceptions class-attribute instance-attribute 

raise_exceptions: bool = False

Whether to raise exceptions from the server.

read_timeout_seconds class-attribute instance-attribute 

read_timeout_seconds: float | None = None

Timeout for read operations.

sampling_callback class-attribute instance-attribute 

sampling_callback: SamplingFnT | None = None

Callback for handling sampling requests.

list_roots_callback class-attribute instance-attribute 

list_roots_callback: ListRootsFnT | None = None

Callback for handling list roots requests.

logging_callback class-attribute instance-attribute 

logging_callback: LoggingFnT | None = None

Callback for handling logging notifications.

message_handler class-attribute instance-attribute 

message_handler: MessageHandlerFnT | None = None

Callback for handling raw messages.

client_info class-attribute instance-attribute 

client_info: Implementation | None = None

Client implementation info to send to server.

elicitation_callback class-attribute instance-attribute 

elicitation_callback: ElicitationFnT | None = None

Callback for handling elicitation requests.

__aenter__ async 

__aenter__() -> Client

Enter the async context manager.

Source code in src/mcp/client/client.py 
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
async def __aenter__(self) -> Client:
    """Enter the async context manager."""
    if self._session is not None:
        raise RuntimeError("Client is already entered; cannot reenter")

    async with AsyncExitStack() as exit_stack:
        read_stream, write_stream = await exit_stack.enter_async_context(self._transport)

        self._session = await exit_stack.enter_async_context(
            ClientSession(
                read_stream=read_stream,
                write_stream=write_stream,
                read_timeout_seconds=self.read_timeout_seconds,
                sampling_callback=self.sampling_callback,
                list_roots_callback=self.list_roots_callback,
                logging_callback=self.logging_callback,
                message_handler=self.message_handler,
                client_info=self.client_info,
                elicitation_callback=self.elicitation_callback,
            )
        )

        await self._session.initialize()

        # Transfer ownership to self for __aexit__ to handle
        self._exit_stack = exit_stack.pop_all()
        return self

__aexit__ async 

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None

Exit the async context manager.

Source code in src/mcp/client/client.py 
138
139
140
141
142
async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
    """Exit the async context manager."""
    if self._exit_stack:  # pragma: no branch
        await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
    self._session = None

session property 

session: ClientSession

Get the underlying ClientSession.

This provides access to the full ClientSession API for advanced use cases.

Raises:

Type Description
RuntimeError

If accessed before entering the context manager.

initialize_result property 

initialize_result: InitializeResult

The server's InitializeResult.

Contains server_info, capabilities, instructions, and the negotiated protocol_version. Raises RuntimeError if accessed outside the context manager.

send_ping async 

send_ping(
    *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a ping request to the server.

Source code in src/mcp/client/client.py 
169
170
171
async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Send a ping request to the server."""
    return await self.session.send_ping(meta=meta)

send_progress_notification async 

send_progress_notification(
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None

Send a progress notification to the server.

Source code in src/mcp/client/client.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None:
    """Send a progress notification to the server."""
    await self.session.send_progress_notification(
        progress_token=progress_token,
        progress=progress,
        total=total,
        message=message,
    )

set_logging_level async 

set_logging_level(
    level: LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None
) -> EmptyResult

Set the logging level on the server.

Source code in src/mcp/client/client.py 
188
189
190
async def set_logging_level(self, level: LoggingLevel, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Set the logging level on the server."""
    return await self.session.set_logging_level(level=level, meta=meta)

list_resources async 

list_resources(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> ListResourcesResult

List available resources from the server.

Source code in src/mcp/client/client.py 
192
193
194
195
196
197
198
199
async def list_resources(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
) -> ListResourcesResult:
    """List available resources from the server."""
    return await self.session.list_resources(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

list_resource_templates async 

list_resource_templates(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> ListResourceTemplatesResult

List available resource templates from the server.

Source code in src/mcp/client/client.py 
201
202
203
204
205
206
207
208
async def list_resource_templates(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
) -> ListResourceTemplatesResult:
    """List available resource templates from the server."""
    return await self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

read_resource async 

read_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> ReadResourceResult

Read a resource from the server.

Parameters:

Name Type Description Default
uri str

The URI of the resource to read.

 required
meta RequestParamsMeta | None

Additional metadata for the request.

 None

Returns:

Type Description
ReadResourceResult

The resource content.
Source code in src/mcp/client/client.py 
210
211
212
213
214
215
216
217
218
219
220
async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> ReadResourceResult:
    """Read a resource from the server.

    Args:
        uri: The URI of the resource to read.
        meta: Additional metadata for the request.

    Returns:
        The resource content.
    """
    return await self.session.read_resource(uri, meta=meta)

subscribe_resource async 

subscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Subscribe to resource updates.

Source code in src/mcp/client/client.py 
222
223
224
async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Subscribe to resource updates."""
    return await self.session.subscribe_resource(uri, meta=meta)

unsubscribe_resource async 

unsubscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Unsubscribe from resource updates.

Source code in src/mcp/client/client.py 
226
227
228
async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Unsubscribe from resource updates."""
    return await self.session.unsubscribe_resource(uri, meta=meta)

call_tool async 

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

Call a tool on the server.

Parameters:

Name Type Description Default
name str

The name of the tool to call

 required
arguments dict[str, Any] | None

Arguments to pass to the tool

 None
read_timeout_seconds float | None

Timeout for the tool call

 None
progress_callback ProgressFnT | None

Callback for progress updates

 None
meta RequestParamsMeta | None

Additional metadata for the request

 None

Returns:

Type Description
CallToolResult

The tool result.
Source code in src/mcp/client/client.py 
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
255
256
257
async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> CallToolResult:
    """Call a tool on the server.

    Args:
        name: The name of the tool to call
        arguments: Arguments to pass to the tool
        read_timeout_seconds: Timeout for the tool call
        progress_callback: Callback for progress updates
        meta: Additional metadata for the request

    Returns:
        The tool result.
    """
    return await self.session.call_tool(
        name=name,
        arguments=arguments,
        read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
        meta=meta,
    )

list_prompts async 

list_prompts(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> ListPromptsResult

List available prompts from the server.

Source code in src/mcp/client/client.py 
259
260
261
262
263
264
265
266
async def list_prompts(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
) -> ListPromptsResult:
    """List available prompts from the server."""
    return await self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

get_prompt async 

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> GetPromptResult

Get a prompt from the server.

Parameters:

Name Type Description Default
name str

The name of the prompt

 required
arguments dict[str, str] | None

Arguments to pass to the prompt

 None
meta RequestParamsMeta | None

Additional metadata for the request

 None

Returns:

Type Description
GetPromptResult

The prompt content.
Source code in src/mcp/client/client.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
async def get_prompt(
    self, name: str, arguments: dict[str, str] | None = None, *, meta: RequestParamsMeta | None = None
) -> GetPromptResult:
    """Get a prompt from the server.

    Args:
        name: The name of the prompt
        arguments: Arguments to pass to the prompt
        meta: Additional metadata for the request

    Returns:
        The prompt content.
    """
    return await self.session.get_prompt(name=name, arguments=arguments, meta=meta)

complete async 

complete(
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult

Get completions for a prompt or resource template argument.

Parameters:

Name Type Description Default
ref ResourceTemplateReference | PromptReference

Reference to the prompt or resource template

 required
argument dict[str, str]

The argument to complete

 required
context_arguments dict[str, str] | None

Additional context arguments

 None

Returns:

Type Description
CompleteResult

Completion suggestions.
Source code in src/mcp/client/client.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
async def complete(
    self,
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult:
    """Get completions for a prompt or resource template argument.

    Args:
        ref: Reference to the prompt or resource template
        argument: The argument to complete
        context_arguments: Additional context arguments

    Returns:
        Completion suggestions.
    """
    return await self.session.complete(ref=ref, argument=argument, context_arguments=context_arguments)

list_tools async 

list_tools(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> ListToolsResult

List available tools from the server.

Source code in src/mcp/client/client.py 
301
302
303
async def list_tools(self, *, cursor: str | None = None, meta: RequestParamsMeta | None = None) -> ListToolsResult:
    """List available tools from the server."""
    return await self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

send_roots_list_changed async 

send_roots_list_changed() -> None

Send a notification that the roots list has changed.

Source code in src/mcp/client/client.py 
305
306
307
308
async def send_roots_list_changed(self) -> None:
    """Send a notification that the roots list has changed."""
    # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
    await self.session.send_roots_list_changed()  # pragma: no cover

ClientRequestContext module-attribute 

ClientRequestContext = RequestContext[ClientSession]

Context for handling incoming requests in a client session.

This context is passed to client-side callbacks (sampling, elicitation, list_roots) when the server sends requests to the client.

Attributes:

Name Type Description
request_id

The unique identifier for this request.
meta

Optional metadata associated with the request.
session

The client session handling this request.

ClientSession

Bases: BaseSession[ClientRequest, ClientNotification, ClientResult, ServerRequest, ServerNotification]

Source code in src/mcp/client/session.py 
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
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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
class ClientSession(
    BaseSession[
        types.ClientRequest,
        types.ClientNotification,
        types.ClientResult,
        types.ServerRequest,
        types.ServerNotification,
    ]
):
    def __init__(
        self,
        read_stream: ReadStream[SessionMessage | Exception],
        write_stream: WriteStream[SessionMessage],
        read_timeout_seconds: float | None = None,
        sampling_callback: SamplingFnT | None = None,
        elicitation_callback: ElicitationFnT | None = None,
        list_roots_callback: ListRootsFnT | None = None,
        logging_callback: LoggingFnT | None = None,
        message_handler: MessageHandlerFnT | None = None,
        client_info: types.Implementation | None = None,
        *,
        sampling_capabilities: types.SamplingCapability | None = None,
        experimental_task_handlers: ExperimentalTaskHandlers | None = None,
    ) -> None:
        super().__init__(read_stream, write_stream, read_timeout_seconds=read_timeout_seconds)
        self._client_info = client_info or DEFAULT_CLIENT_INFO
        self._sampling_callback = sampling_callback or _default_sampling_callback
        self._sampling_capabilities = sampling_capabilities
        self._elicitation_callback = elicitation_callback or _default_elicitation_callback
        self._list_roots_callback = list_roots_callback or _default_list_roots_callback
        self._logging_callback = logging_callback or _default_logging_callback
        self._message_handler = message_handler or _default_message_handler
        self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
        self._initialize_result: types.InitializeResult | None = None
        self._experimental_features: ExperimentalClientFeatures | None = None

        # Experimental: Task handlers (use defaults if not provided)
        self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers()

    @property
    def _receive_request_adapter(self) -> TypeAdapter[types.ServerRequest]:
        return types.server_request_adapter

    @property
    def _receive_notification_adapter(self) -> TypeAdapter[types.ServerNotification]:
        return types.server_notification_adapter

    async def initialize(self) -> types.InitializeResult:
        sampling = (
            (self._sampling_capabilities or types.SamplingCapability())
            if self._sampling_callback is not _default_sampling_callback
            else None
        )
        elicitation = (
            types.ElicitationCapability(form=types.FormElicitationCapability(), url=types.UrlElicitationCapability())
            if self._elicitation_callback is not _default_elicitation_callback
            else None
        )
        roots = (
            # TODO: Should this be based on whether we
            # _will_ send notifications, or only whether
            # they're supported?
            types.RootsCapability(list_changed=True)
            if self._list_roots_callback is not _default_list_roots_callback
            else None
        )

        result = await self.send_request(
            types.InitializeRequest(
                params=types.InitializeRequestParams(
                    protocol_version=types.LATEST_PROTOCOL_VERSION,
                    capabilities=types.ClientCapabilities(
                        sampling=sampling,
                        elicitation=elicitation,
                        experimental=None,
                        roots=roots,
                        tasks=self._task_handlers.build_capability(),
                    ),
                    client_info=self._client_info,
                ),
            ),
            types.InitializeResult,
        )

        if result.protocol_version not in SUPPORTED_PROTOCOL_VERSIONS:
            raise RuntimeError(f"Unsupported protocol version from the server: {result.protocol_version}")

        self._initialize_result = result

        await self.send_notification(types.InitializedNotification())

        return result

    @property
    def initialize_result(self) -> types.InitializeResult | None:
        """The server's InitializeResult. None until initialize() has been called.

        Contains server_info, capabilities, instructions, and the negotiated protocol_version.
        """
        return self._initialize_result

    @property
    def experimental(self) -> ExperimentalClientFeatures:
        """Experimental APIs for tasks and other features.

        !!! warning
            These APIs are experimental and may change without notice.

        Example:
            ```python
            status = await session.experimental.get_task(task_id)
            result = await session.experimental.get_task_result(task_id, CallToolResult)
            ```
        """
        if self._experimental_features is None:
            self._experimental_features = ExperimentalClientFeatures(self)
        return self._experimental_features

    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a ping request."""
        return await self.send_request(types.PingRequest(params=types.RequestParams(_meta=meta)), types.EmptyResult)

    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> None:
        """Send a progress notification."""
        await self.send_notification(
            types.ProgressNotification(
                params=types.ProgressNotificationParams(
                    progress_token=progress_token,
                    progress=progress,
                    total=total,
                    message=message,
                    _meta=meta,
                ),
            )
        )

    async def set_logging_level(
        self,
        level: types.LoggingLevel,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> types.EmptyResult:
        """Send a logging/setLevel request."""
        return await self.send_request(
            types.SetLevelRequest(params=types.SetLevelRequestParams(level=level, _meta=meta)),
            types.EmptyResult,
        )

    async def list_resources(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListResourcesResult:
        """Send a resources/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(types.ListResourcesRequest(params=params), types.ListResourcesResult)

    async def list_resource_templates(
        self, *, params: types.PaginatedRequestParams | None = None
    ) -> types.ListResourceTemplatesResult:
        """Send a resources/templates/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(
            types.ListResourceTemplatesRequest(params=params),
            types.ListResourceTemplatesResult,
        )

    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
        """Send a resources/read request."""
        return await self.send_request(
            types.ReadResourceRequest(params=types.ReadResourceRequestParams(uri=uri, _meta=meta)),
            types.ReadResourceResult,
        )

    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a resources/subscribe request."""
        return await self.send_request(
            types.SubscribeRequest(params=types.SubscribeRequestParams(uri=uri, _meta=meta)),
            types.EmptyResult,
        )

    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a resources/unsubscribe request."""
        return await self.send_request(
            types.UnsubscribeRequest(params=types.UnsubscribeRequestParams(uri=uri, _meta=meta)),
            types.EmptyResult,
        )

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> types.CallToolResult:
        """Send a tools/call request with optional progress callback support."""

        result = await self.send_request(
            types.CallToolRequest(
                params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
            ),
            types.CallToolResult,
            request_read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
        )

        if not result.is_error:
            await self._validate_tool_result(name, result)

        return result

    async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
        """Validate the structured content of a tool result against its output schema."""
        if name not in self._tool_output_schemas:
            # refresh output schema cache
            await self.list_tools()

        output_schema = None
        if name in self._tool_output_schemas:
            output_schema = self._tool_output_schemas.get(name)
        else:
            logger.warning(f"Tool {name} not listed by server, cannot validate any structured content")

        if output_schema is not None:
            from jsonschema import SchemaError, ValidationError, validate

            if result.structured_content is None:
                raise RuntimeError(
                    f"Tool {name} has an output schema but did not return structured content"
                )  # pragma: no cover
            try:
                validate(result.structured_content, output_schema)
            except ValidationError as e:
                raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
            except SchemaError as e:  # pragma: no cover
                raise RuntimeError(f"Invalid schema for tool {name}: {e}")  # pragma: no cover

    async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
        """Send a prompts/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)

    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> types.GetPromptResult:
        """Send a prompts/get request."""
        return await self.send_request(
            types.GetPromptRequest(params=types.GetPromptRequestParams(name=name, arguments=arguments, _meta=meta)),
            types.GetPromptResult,
        )

    async def complete(
        self,
        ref: types.ResourceTemplateReference | types.PromptReference,
        argument: dict[str, str],
        context_arguments: dict[str, str] | None = None,
    ) -> types.CompleteResult:
        """Send a completion/complete request."""
        context = None
        if context_arguments is not None:
            context = types.CompletionContext(arguments=context_arguments)

        return await self.send_request(
            types.CompleteRequest(
                params=types.CompleteRequestParams(
                    ref=ref,
                    argument=types.CompletionArgument(**argument),
                    context=context,
                ),
            ),
            types.CompleteResult,
        )

    async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult:
        """Send a tools/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        result = await self.send_request(
            types.ListToolsRequest(params=params),
            types.ListToolsResult,
        )

        # Cache tool output schemas for future validation
        # Note: don't clear the cache, as we may be using a cursor
        for tool in result.tools:
            self._tool_output_schemas[tool.name] = tool.output_schema

        return result

    async def send_roots_list_changed(self) -> None:  # pragma: no cover
        """Send a roots/list_changed notification."""
        await self.send_notification(types.RootsListChangedNotification())

    async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
        ctx = RequestContext[ClientSession](request_id=responder.request_id, meta=responder.request_meta, session=self)

        # Delegate to experimental task handler if applicable
        if self._task_handlers.handles_request(responder.request):
            with responder:
                await self._task_handlers.handle_request(ctx, responder)
            return None

        # Core request handling
        match responder.request:
            case types.CreateMessageRequest(params=params):
                with responder:
                    # Check if this is a task-augmented request
                    if params.task is not None:
                        response = await self._task_handlers.augmented_sampling(ctx, params, params.task)
                    else:
                        response = await self._sampling_callback(ctx, params)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.ElicitRequest(params=params):
                with responder:
                    # Check if this is a task-augmented request
                    if params.task is not None:
                        response = await self._task_handlers.augmented_elicitation(ctx, params, params.task)
                    else:
                        response = await self._elicitation_callback(ctx, params)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.ListRootsRequest():
                with responder:
                    response = await self._list_roots_callback(ctx)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.PingRequest():  # pragma: no cover
                with responder:
                    return await responder.respond(types.EmptyResult())

            case _:  # pragma: no cover
                pass  # Task requests handled above by _task_handlers

        return None

    async def _handle_incoming(
        self,
        req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
    ) -> None:
        """Handle incoming messages by forwarding to the message handler."""
        await self._message_handler(req)

    async def _received_notification(self, notification: types.ServerNotification) -> None:
        """Handle notifications from the server."""
        # Process specific notification types
        match notification:
            case types.LoggingMessageNotification(params=params):
                await self._logging_callback(params)
            case types.ElicitCompleteNotification(params=params):
                # Handle elicitation completion notification
                # Clients MAY use this to retry requests or update UI
                # The notification contains the elicitationId of the completed elicitation
                pass
            case _:
                pass

initialize_result property 

initialize_result: InitializeResult | None

The server's InitializeResult. None until initialize() has been called.

Contains server_info, capabilities, instructions, and the negotiated protocol_version.

experimental property 

experimental: ExperimentalClientFeatures

Experimental APIs for tasks and other features.

Warning

These APIs are experimental and may change without notice.

Example 
status = await session.experimental.get_task(task_id)
result = await session.experimental.get_task_result(task_id, CallToolResult)

send_ping async 

send_ping(
    *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a ping request.

Source code in src/mcp/client/session.py 
219
220
221
async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a ping request."""
    return await self.send_request(types.PingRequest(params=types.RequestParams(_meta=meta)), types.EmptyResult)

send_progress_notification async 

send_progress_notification(
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> None

Send a progress notification.

Source code in src/mcp/client/session.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> None:
    """Send a progress notification."""
    await self.send_notification(
        types.ProgressNotification(
            params=types.ProgressNotificationParams(
                progress_token=progress_token,
                progress=progress,
                total=total,
                message=message,
                _meta=meta,
            ),
        )
    )

set_logging_level async 

set_logging_level(
    level: LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a logging/setLevel request.

Source code in src/mcp/client/session.py 
245
246
247
248
249
250
251
252
253
254
255
async def set_logging_level(
    self,
    level: types.LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.EmptyResult:
    """Send a logging/setLevel request."""
    return await self.send_request(
        types.SetLevelRequest(params=types.SetLevelRequestParams(level=level, _meta=meta)),
        types.EmptyResult,
    )

list_resources async 

list_resources(
    *, params: PaginatedRequestParams | None = None
) -> ListResourcesResult

Send a resources/list request.

Parameters:

Name Type Description Default
params PaginatedRequestParams | None

Full pagination parameters including cursor and any future fields

 None
Source code in src/mcp/client/session.py 
257
258
259
260
261
262
263
async def list_resources(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListResourcesResult:
    """Send a resources/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(types.ListResourcesRequest(params=params), types.ListResourcesResult)

list_resource_templates async 

list_resource_templates(
    *, params: PaginatedRequestParams | None = None
) -> ListResourceTemplatesResult

Send a resources/templates/list request.

Parameters:

Name Type Description Default
params PaginatedRequestParams | None

Full pagination parameters including cursor and any future fields

 None
Source code in src/mcp/client/session.py 
265
266
267
268
269
270
271
272
273
274
275
276
async def list_resource_templates(
    self, *, params: types.PaginatedRequestParams | None = None
) -> types.ListResourceTemplatesResult:
    """Send a resources/templates/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(
        types.ListResourceTemplatesRequest(params=params),
        types.ListResourceTemplatesResult,
    )

read_resource async 

read_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> ReadResourceResult

Send a resources/read request.

Source code in src/mcp/client/session.py 
278
279
280
281
282
283
async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
    """Send a resources/read request."""
    return await self.send_request(
        types.ReadResourceRequest(params=types.ReadResourceRequestParams(uri=uri, _meta=meta)),
        types.ReadResourceResult,
    )

subscribe_resource async 

subscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a resources/subscribe request.

Source code in src/mcp/client/session.py 
285
286
287
288
289
290
async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a resources/subscribe request."""
    return await self.send_request(
        types.SubscribeRequest(params=types.SubscribeRequestParams(uri=uri, _meta=meta)),
        types.EmptyResult,
    )

unsubscribe_resource async 

unsubscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a resources/unsubscribe request.

Source code in src/mcp/client/session.py 
292
293
294
295
296
297
async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a resources/unsubscribe request."""
    return await self.send_request(
        types.UnsubscribeRequest(params=types.UnsubscribeRequestParams(uri=uri, _meta=meta)),
        types.EmptyResult,
    )

call_tool async 

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

Send a tools/call request with optional progress callback support.

Source code in src/mcp/client/session.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.CallToolResult:
    """Send a tools/call request with optional progress callback support."""

    result = await self.send_request(
        types.CallToolRequest(
            params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
        ),
        types.CallToolResult,
        request_read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
    )

    if not result.is_error:
        await self._validate_tool_result(name, result)

    return result

list_prompts async 

list_prompts(
    *, params: PaginatedRequestParams | None = None
) -> ListPromptsResult

Send a prompts/list request.

Parameters:

Name Type Description Default
params PaginatedRequestParams | None

Full pagination parameters including cursor and any future fields

 None
Source code in src/mcp/client/session.py 
350
351
352
353
354
355
356
async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
    """Send a prompts/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)

get_prompt async 

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> GetPromptResult

Send a prompts/get request.

Source code in src/mcp/client/session.py 
358
359
360
361
362
363
364
365
366
367
368
369
async def get_prompt(
    self,
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.GetPromptResult:
    """Send a prompts/get request."""
    return await self.send_request(
        types.GetPromptRequest(params=types.GetPromptRequestParams(name=name, arguments=arguments, _meta=meta)),
        types.GetPromptResult,
    )

complete async 

complete(
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult

Send a completion/complete request.

Source code in src/mcp/client/session.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
async def complete(
    self,
    ref: types.ResourceTemplateReference | types.PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> types.CompleteResult:
    """Send a completion/complete request."""
    context = None
    if context_arguments is not None:
        context = types.CompletionContext(arguments=context_arguments)

    return await self.send_request(
        types.CompleteRequest(
            params=types.CompleteRequestParams(
                ref=ref,
                argument=types.CompletionArgument(**argument),
                context=context,
            ),
        ),
        types.CompleteResult,
    )

list_tools async 

list_tools(
    *, params: PaginatedRequestParams | None = None
) -> ListToolsResult

Send a tools/list request.

Parameters:

Name Type Description Default
params PaginatedRequestParams | None

Full pagination parameters including cursor and any future fields

 None
Source code in src/mcp/client/session.py 
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult:
    """Send a tools/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    result = await self.send_request(
        types.ListToolsRequest(params=params),
        types.ListToolsResult,
    )

    # Cache tool output schemas for future validation
    # Note: don't clear the cache, as we may be using a cursor
    for tool in result.tools:
        self._tool_output_schemas[tool.name] = tool.output_schema

    return result

send_roots_list_changed async 

send_roots_list_changed() -> None

Send a roots/list_changed notification.

Source code in src/mcp/client/session.py 
411
412
413
async def send_roots_list_changed(self) -> None:  # pragma: no cover
    """Send a roots/list_changed notification."""
    await self.send_notification(types.RootsListChangedNotification())