mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 21:14:17 +02:00
Merge pull request #301 from jlowin/client
Add low-level methods to client
This commit is contained in:
Jeremiah Lowin
GitHub
commit
c2d5023041
6 changed files with 383 additions and 48 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,190 @@ 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
|
||||
# --- Call Tool ---
|
||||
|
||||
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: Literal[False] = False,
|
||||
) -> list[
|
||||
mcp.types.TextContent | mcp.types.ImageContent | mcp.types.EmbeddedResource
|
||||
]: ...
|
||||
]:
|
||||
"""Call a tool on the server.
|
||||
|
||||
@overload
|
||||
async def call_tool(
|
||||
self,
|
||||
name: str,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
_return_raw_result: Literal[True] = True,
|
||||
) -> mcp.types.CallToolResult: ...
|
||||
Unlike call_tool_mcp, this method raises a ClientError if the tool call results in an error.
|
||||
|
||||
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:
|
||||
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
|
||||
|
|
|
|||
|
|
@ -123,9 +123,7 @@ class BulkToolCaller(MCPMixin):
|
|||
"""
|
||||
|
||||
async with Client(self.connection) as client:
|
||||
result = await client.call_tool(
|
||||
name=tool, arguments=arguments, _return_raw_result=True
|
||||
)
|
||||
result = await client.call_tool_mcp(name=tool, arguments=arguments)
|
||||
|
||||
return CallToolRequestResult(
|
||||
tool=tool,
|
||||
|
|
|
|||
|
|
@ -65,8 +65,9 @@ class ProxyTool(Tool):
|
|||
# the client context manager will swallow any exceptions inside a TaskGroup
|
||||
# so we return the raw result and raise an exception ourselves
|
||||
async with self._client:
|
||||
result = await self._client.call_tool(
|
||||
self.name, arguments, _return_raw_result=True
|
||||
result = await self._client.call_tool_mcp(
|
||||
name=self.name,
|
||||
arguments=arguments,
|
||||
)
|
||||
if result.isError:
|
||||
raise ValueError(cast(mcp.types.TextContent, result.content[0]).text)
|
||||
|
|
|
|||
|
|
@ -79,6 +79,19 @@ async def test_list_tools(fastmcp_server):
|
|||
assert set(tool.name for tool in result) == {"greet", "add"}
|
||||
|
||||
|
||||
async def test_list_tools_mcp(fastmcp_server):
|
||||
"""Test the list_tools_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.list_tools_mcp()
|
||||
|
||||
# Check that we got the raw MCP ListToolsResult object
|
||||
assert hasattr(result, "tools")
|
||||
assert len(result.tools) == 2
|
||||
assert set(tool.name for tool in result.tools) == {"greet", "add"}
|
||||
|
||||
|
||||
async def test_call_tool(fastmcp_server):
|
||||
"""Test calling a tool with InMemoryClient."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -91,6 +104,26 @@ async def test_call_tool(fastmcp_server):
|
|||
assert "Hello, World!" in content_str
|
||||
|
||||
|
||||
async def test_call_tool_mcp(fastmcp_server):
|
||||
"""Test the call_tool_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.call_tool_mcp("greet", {"name": "World"})
|
||||
|
||||
# Check that we got the raw MCP CallToolResult object
|
||||
assert hasattr(result, "content")
|
||||
assert hasattr(result, "isError")
|
||||
assert result.isError is False
|
||||
# The content is a list, so we'll check the first element
|
||||
# by properly accessing it
|
||||
content = result.content
|
||||
assert len(content) > 0
|
||||
first_content = content[0]
|
||||
content_str = str(first_content)
|
||||
assert "Hello, World!" in content_str
|
||||
|
||||
|
||||
async def test_list_resources(fastmcp_server):
|
||||
"""Test listing resources with InMemoryClient."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -103,6 +136,19 @@ async def test_list_resources(fastmcp_server):
|
|||
assert str(result[0].uri) == "data://users"
|
||||
|
||||
|
||||
async def test_list_resources_mcp(fastmcp_server):
|
||||
"""Test the list_resources_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.list_resources_mcp()
|
||||
|
||||
# Check that we got the raw MCP ListResourcesResult object
|
||||
assert hasattr(result, "resources")
|
||||
assert len(result.resources) == 1
|
||||
assert str(result.resources[0].uri) == "data://users"
|
||||
|
||||
|
||||
async def test_list_prompts(fastmcp_server):
|
||||
"""Test listing prompts with InMemoryClient."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -115,6 +161,19 @@ async def test_list_prompts(fastmcp_server):
|
|||
assert result[0].name == "welcome"
|
||||
|
||||
|
||||
async def test_list_prompts_mcp(fastmcp_server):
|
||||
"""Test the list_prompts_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.list_prompts_mcp()
|
||||
|
||||
# Check that we got the raw MCP ListPromptsResult object
|
||||
assert hasattr(result, "prompts")
|
||||
assert len(result.prompts) == 1
|
||||
assert result.prompts[0].name == "welcome"
|
||||
|
||||
|
||||
async def test_get_prompt(fastmcp_server):
|
||||
"""Test getting a prompt with InMemoryClient."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -127,6 +186,20 @@ async def test_get_prompt(fastmcp_server):
|
|||
assert "Welcome to FastMCP, Developer!" in result_str
|
||||
|
||||
|
||||
async def test_get_prompt_mcp(fastmcp_server):
|
||||
"""Test the get_prompt_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.get_prompt_mcp("welcome", {"name": "Developer"})
|
||||
|
||||
# Check that we got the raw MCP GetPromptResult object
|
||||
assert hasattr(result, "messages")
|
||||
assert len(result.messages) > 0
|
||||
result_str = str(result.messages)
|
||||
assert "Welcome to FastMCP, Developer!" in result_str
|
||||
|
||||
|
||||
async def test_read_resource(fastmcp_server):
|
||||
"""Test reading a resource with InMemoryClient."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -145,6 +218,26 @@ async def test_read_resource(fastmcp_server):
|
|||
assert "Charlie" in contents_str
|
||||
|
||||
|
||||
async def test_read_resource_mcp(fastmcp_server):
|
||||
"""Test the read_resource_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
# Use the URI from the resource we know exists in our server
|
||||
uri = cast(
|
||||
AnyUrl, "data://users"
|
||||
) # Use cast for type hint only, the URI is valid
|
||||
result = await client.read_resource_mcp(uri)
|
||||
|
||||
# Check that we got the raw MCP ReadResourceResult object
|
||||
assert hasattr(result, "contents")
|
||||
assert len(result.contents) > 0
|
||||
contents_str = str(result.contents[0])
|
||||
assert "Alice" in contents_str
|
||||
assert "Bob" in contents_str
|
||||
assert "Charlie" in contents_str
|
||||
|
||||
|
||||
async def test_client_connection(fastmcp_server):
|
||||
"""Test that the client connects and disconnects properly."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
@ -213,6 +306,19 @@ async def test_resource_template(fastmcp_server):
|
|||
assert '"active": true' in content_str
|
||||
|
||||
|
||||
async def test_list_resource_templates_mcp(fastmcp_server):
|
||||
"""Test the list_resource_templates_mcp method that returns raw MCP protocol objects."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
||||
async with client:
|
||||
result = await client.list_resource_templates_mcp()
|
||||
|
||||
# Check that we got the raw MCP ListResourceTemplatesResult object
|
||||
assert hasattr(result, "resourceTemplates")
|
||||
assert len(result.resourceTemplates) == 1
|
||||
assert "data://user/{user_id}" in result.resourceTemplates[0].uriTemplate
|
||||
|
||||
|
||||
async def test_mcp_resource_generation(fastmcp_server):
|
||||
"""Test that resources are properly generated in MCP format."""
|
||||
client = Client(transport=FastMCPTransport(fastmcp_server))
|
||||
|
|
|
|||
|
|
@ -96,7 +96,7 @@ class TestTools:
|
|||
|
||||
async def test_call_tool_error_as_client_raw(self, tool_server: FastMCP):
|
||||
async with Client(tool_server) as client:
|
||||
result = await client.call_tool("error_tool", {}, _return_raw_result=True)
|
||||
result = await client.call_tool_mcp("error_tool", {})
|
||||
assert result.isError
|
||||
assert isinstance(result.content[0], TextContent)
|
||||
assert "Test error" in result.content[0].text
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue