mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 04:54:17 +02:00
add _mcp methods
This commit is contained in:
Jeremiah Lowin
parent
11cf3af559
commit
cebe4614bb
2 changed files with 271 additions and 38 deletions
|
|
@ -149,6 +149,34 @@ The `Client` provides methods corresponding to standard MCP requests:
|
|||
* **`list_prompts()`**: Retrieves available prompt templates.
|
||||
* **`get_prompt(name: str, arguments: dict[str, Any] | None = None)`**: Retrieves a rendered prompt message list.
|
||||
|
||||
### Raw MCP Protocol Objects
|
||||
|
||||
The FastMCP client attempts to provide a "friendly" interface to the MCP protocol, but sometimes you may need access to the raw MCP protocol objects. Each of the main client methods that returns data has a corresponding `*_mcp` method that returns the raw MCP protocol objects directly.
|
||||
|
||||
```python
|
||||
# Standard method - returns just the list of tools
|
||||
tools = await client.list_tools()
|
||||
# tools -> list[mcp.types.Tool]
|
||||
|
||||
# Raw MCP method - returns the full protocol object
|
||||
result = await client.list_tools_mcp()
|
||||
# result -> mcp.types.ListToolsResult
|
||||
tools = result.tools
|
||||
```
|
||||
|
||||
Available raw MCP methods:
|
||||
|
||||
* **`list_tools_mcp()`**: Returns `mcp.types.ListToolsResult`
|
||||
* **`call_tool_mcp(name, arguments)`**: Returns `mcp.types.CallToolResult`
|
||||
* **`list_resources_mcp()`**: Returns `mcp.types.ListResourcesResult`
|
||||
* **`list_resource_templates_mcp()`**: Returns `mcp.types.ListResourceTemplatesResult`
|
||||
* **`read_resource_mcp(uri)`**: Returns `mcp.types.ReadResourceResult`
|
||||
* **`list_prompts_mcp()`**: Returns `mcp.types.ListPromptsResult`
|
||||
* **`get_prompt_mcp(name, arguments)`**: Returns `mcp.types.GetPromptResult`
|
||||
* **`complete_mcp(ref, argument)`**: Returns `mcp.types.CompleteResult`
|
||||
|
||||
These methods are especially useful for debugging or when you need to access metadata or fields that aren't exposed by the simplified methods.
|
||||
|
||||
### Advanced Features
|
||||
|
||||
MCP allows servers to interact with clients in order to provide additional capabilities. The `Client` constructor accepts additional configuration to handle these server requests.
|
||||
|
|
@ -268,5 +296,5 @@ async def safe_call_tool():
|
|||
Other errors, like connection failures, will raise standard Python exceptions (e.g., `ConnectionError`, `TimeoutError`).
|
||||
|
||||
<Tip>
|
||||
The client transport often has its own error-handling mechanisms, so you can not always trap errors like those raised by `call_tool` outside of the `async with` block. Instead, you can call `call_tool(..., _return_raw_result=True)` to get the raw `mcp.types.CallToolResult` object and handle errors yourself by checking its `isError` attribute.
|
||||
The client transport often has its own error-handling mechanisms, so you can not always trap errors like those raised by `call_tool` outside of the `async with` block. Instead, you can use `call_tool_mcp()` to get the raw `mcp.types.CallToolResult` object and handle errors yourself by checking its `isError` attribute.
|
||||
</Tip>
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import datetime
|
||||
from contextlib import AbstractAsyncContextManager
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal, cast, overload
|
||||
from typing import Any, cast
|
||||
|
||||
import mcp.types
|
||||
from mcp import ClientSession
|
||||
|
|
@ -107,6 +107,7 @@ class Client:
|
|||
self._session = None
|
||||
|
||||
# --- MCP Client Methods ---
|
||||
|
||||
async def ping(self) -> None:
|
||||
"""Send a ping request."""
|
||||
await self.session.send_ping()
|
||||
|
|
@ -128,23 +129,100 @@ class Client:
|
|||
"""Send a roots/list_changed notification."""
|
||||
await self.session.send_roots_list_changed()
|
||||
|
||||
async def list_resources(self) -> list[mcp.types.Resource]:
|
||||
"""Send a resources/list request."""
|
||||
# --- Resources ---
|
||||
|
||||
async def list_resources_mcp(self) -> mcp.types.ListResourcesResult:
|
||||
"""Send a resources/list request and return the complete MCP protocol result.
|
||||
|
||||
Returns:
|
||||
mcp.types.ListResourcesResult: The complete response object from the protocol,
|
||||
containing the list of resources and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.list_resources()
|
||||
return result
|
||||
|
||||
async def list_resources(self) -> list[mcp.types.Resource]:
|
||||
"""Retrieve a list of resources available on the server.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.Resource]: A list of Resource objects.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.list_resources_mcp()
|
||||
return result.resources
|
||||
|
||||
async def list_resource_templates(self) -> list[mcp.types.ResourceTemplate]:
|
||||
"""Send a resources/listResourceTemplates request."""
|
||||
async def list_resource_templates_mcp(
|
||||
self,
|
||||
) -> mcp.types.ListResourceTemplatesResult:
|
||||
"""Send a resources/listResourceTemplates request and return the complete MCP protocol result.
|
||||
|
||||
Returns:
|
||||
mcp.types.ListResourceTemplatesResult: The complete response object from the protocol,
|
||||
containing the list of resource templates and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.list_resource_templates()
|
||||
return result
|
||||
|
||||
async def list_resource_templates(
|
||||
self,
|
||||
) -> list[mcp.types.ResourceTemplate]:
|
||||
"""Retrieve a list of resource templates available on the server.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.ResourceTemplate]: A list of ResourceTemplate objects.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.list_resource_templates_mcp()
|
||||
return result.resourceTemplates
|
||||
|
||||
async def read_resource_mcp(
|
||||
self, uri: AnyUrl | str
|
||||
) -> mcp.types.ReadResourceResult:
|
||||
"""Send a resources/read request and return the complete MCP protocol result.
|
||||
|
||||
Args:
|
||||
uri (AnyUrl | str): The URI of the resource to read. Can be a string or an AnyUrl object.
|
||||
|
||||
Returns:
|
||||
mcp.types.ReadResourceResult: The complete response object from the protocol,
|
||||
containing the resource contents and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
if isinstance(uri, str):
|
||||
uri = AnyUrl(uri) # Ensure AnyUrl
|
||||
result = await self.session.read_resource(uri)
|
||||
return result
|
||||
|
||||
async def read_resource(
|
||||
self, uri: AnyUrl | str
|
||||
) -> list[mcp.types.TextResourceContents | mcp.types.BlobResourceContents]:
|
||||
"""Send a resources/read request."""
|
||||
"""Read the contents of a resource or resolved template.
|
||||
|
||||
Args:
|
||||
uri (AnyUrl | str): The URI of the resource to read. Can be a string or an AnyUrl object.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.TextResourceContents | mcp.types.BlobResourceContents]: A list of content
|
||||
objects, typically containing either text or binary data.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
if isinstance(uri, str):
|
||||
uri = AnyUrl(uri) # Ensure AnyUrl
|
||||
result = await self.session.read_resource(uri)
|
||||
result = await self.read_resource_mcp(uri)
|
||||
return result.contents
|
||||
|
||||
# async def subscribe_resource(self, uri: AnyUrl | str) -> None:
|
||||
|
|
@ -159,66 +237,193 @@ class Client:
|
|||
# uri = AnyUrl(uri)
|
||||
# await self.session.unsubscribe_resource(uri)
|
||||
|
||||
async def list_prompts(self) -> list[mcp.types.Prompt]:
|
||||
"""Send a prompts/list request."""
|
||||
# --- Prompts ---
|
||||
|
||||
async def list_prompts_mcp(self) -> mcp.types.ListPromptsResult:
|
||||
"""Send a prompts/list request and return the complete MCP protocol result.
|
||||
|
||||
Returns:
|
||||
mcp.types.ListPromptsResult: The complete response object from the protocol,
|
||||
containing the list of prompts and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.list_prompts()
|
||||
return result
|
||||
|
||||
async def list_prompts(self) -> list[mcp.types.Prompt]:
|
||||
"""Retrieve a list of prompts available on the server.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.Prompt]: A list of Prompt objects.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.list_prompts_mcp()
|
||||
return result.prompts
|
||||
|
||||
# --- Prompt ---
|
||||
async def get_prompt_mcp(
|
||||
self, name: str, arguments: dict[str, str] | None = None
|
||||
) -> mcp.types.GetPromptResult:
|
||||
"""Send a prompts/get request and return the complete MCP protocol result.
|
||||
|
||||
Args:
|
||||
name (str): The name of the prompt to retrieve.
|
||||
arguments (dict[str, str] | None, optional): Arguments to pass to the prompt. Defaults to None.
|
||||
|
||||
Returns:
|
||||
mcp.types.GetPromptResult: The complete response object from the protocol,
|
||||
containing the prompt messages and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.get_prompt(name=name, arguments=arguments)
|
||||
return result
|
||||
|
||||
async def get_prompt(
|
||||
self, name: str, arguments: dict[str, str] | None = None
|
||||
) -> list[mcp.types.PromptMessage]:
|
||||
"""Send a prompts/get request."""
|
||||
result = await self.session.get_prompt(name, arguments)
|
||||
"""Retrieve a rendered prompt message list from the server.
|
||||
|
||||
Args:
|
||||
name (str): The name of the prompt to retrieve.
|
||||
arguments (dict[str, str] | None, optional): Arguments to pass to the prompt. Defaults to None.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.PromptMessage]: A list of prompt messages.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.get_prompt_mcp(name=name, arguments=arguments)
|
||||
return result.messages
|
||||
|
||||
# --- Completion ---
|
||||
|
||||
async def complete_mcp(
|
||||
self,
|
||||
ref: mcp.types.ResourceReference | mcp.types.PromptReference,
|
||||
argument: dict[str, str],
|
||||
) -> mcp.types.CompleteResult:
|
||||
"""Send a completion request and return the complete MCP protocol result.
|
||||
|
||||
Args:
|
||||
ref (mcp.types.ResourceReference | mcp.types.PromptReference): The reference to complete.
|
||||
argument (dict[str, str]): Arguments to pass to the completion request.
|
||||
|
||||
Returns:
|
||||
mcp.types.CompleteResult: The complete response object from the protocol,
|
||||
containing the completion and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.complete(ref=ref, argument=argument)
|
||||
return result
|
||||
|
||||
async def complete(
|
||||
self,
|
||||
ref: mcp.types.ResourceReference | mcp.types.PromptReference,
|
||||
argument: dict[str, str],
|
||||
) -> mcp.types.Completion:
|
||||
"""Send a completion request."""
|
||||
result = await self.session.complete(ref, argument)
|
||||
"""Send a completion request to the server.
|
||||
|
||||
Args:
|
||||
ref (mcp.types.ResourceReference | mcp.types.PromptReference): The reference to complete.
|
||||
argument (dict[str, str]): Arguments to pass to the completion request.
|
||||
|
||||
Returns:
|
||||
mcp.types.Completion: The completion object.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.complete_mcp(ref=ref, argument=argument)
|
||||
return result.completion
|
||||
|
||||
async def list_tools(self) -> list[mcp.types.Tool]:
|
||||
"""Send a tools/list request."""
|
||||
# --- Tools ---
|
||||
|
||||
async def list_tools_mcp(self) -> mcp.types.ListToolsResult:
|
||||
"""Send a tools/list request and return the complete MCP protocol result.
|
||||
|
||||
Returns:
|
||||
mcp.types.ListToolsResult: The complete response object from the protocol,
|
||||
containing the list of tools and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.list_tools()
|
||||
return result
|
||||
|
||||
async def list_tools(self) -> list[mcp.types.Tool]:
|
||||
"""Retrieve a list of tools available on the server.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.Tool]: A list of Tool objects.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.list_tools_mcp()
|
||||
return result.tools
|
||||
|
||||
@overload
|
||||
async def call_tool(
|
||||
self,
|
||||
name: str,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
_return_raw_result: Literal[False] = False,
|
||||
) -> list[
|
||||
mcp.types.TextContent | mcp.types.ImageContent | mcp.types.EmbeddedResource
|
||||
]: ...
|
||||
# --- Call Tool ---
|
||||
|
||||
@overload
|
||||
async def call_tool(
|
||||
self,
|
||||
name: str,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
_return_raw_result: Literal[True] = True,
|
||||
) -> mcp.types.CallToolResult: ...
|
||||
async def call_tool_mcp(
|
||||
self, name: str, arguments: dict[str, Any]
|
||||
) -> mcp.types.CallToolResult:
|
||||
"""Send a tools/call request and return the complete MCP protocol result.
|
||||
|
||||
This method returns the raw CallToolResult object, which includes an isError flag
|
||||
and other metadata. It does not raise an exception if the tool call results in an error.
|
||||
|
||||
Args:
|
||||
name (str): The name of the tool to call.
|
||||
arguments (dict[str, Any]): Arguments to pass to the tool.
|
||||
|
||||
Returns:
|
||||
mcp.types.CallToolResult: The complete response object from the protocol,
|
||||
containing the tool result and any additional metadata.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.session.call_tool(name=name, arguments=arguments)
|
||||
return result
|
||||
|
||||
async def call_tool(
|
||||
self,
|
||||
name: str,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
_return_raw_result: bool = False,
|
||||
) -> (
|
||||
list[
|
||||
mcp.types.TextContent | mcp.types.ImageContent | mcp.types.EmbeddedResource
|
||||
]
|
||||
| mcp.types.CallToolResult
|
||||
):
|
||||
"""Send a tools/call request."""
|
||||
result = await self.session.call_tool(name, arguments)
|
||||
if _return_raw_result:
|
||||
return result
|
||||
elif result.isError:
|
||||
"""Call a tool on the server.
|
||||
|
||||
Unlike call_tool_mcp, this method raises a ClientError if the tool call results in an error.
|
||||
|
||||
Args:
|
||||
name (str): The name of the tool to call.
|
||||
arguments (dict[str, Any] | None, optional): Arguments to pass to the tool. Defaults to None.
|
||||
|
||||
Returns:
|
||||
list[mcp.types.TextContent | mcp.types.ImageContent | mcp.types.EmbeddedResource]:
|
||||
The content returned by the tool.
|
||||
|
||||
Raises:
|
||||
ClientError: If the tool call results in an error.
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.call_tool_mcp(name=name, arguments=arguments or {})
|
||||
if result.isError:
|
||||
msg = cast(mcp.types.TextContent, result.content[0]).text
|
||||
raise ClientError(msg)
|
||||
return result.content
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue