ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 21:44:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #284 from jlowin/client-docs

Browse source 
Fix client docs for advanced features, add tests for logging
This commit is contained in:
Jeremiah Lowin 2025-04-30 12:10:21 -04:00 • committed by GitHub
commit 056a3bff3b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 175 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

131
docs/clients/client.mdx
View file

@ -149,83 +149,90 @@ 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.
### Callbacks
### Advanced Features
MCP allows servers to make requests *back* to the client for certain capabilities. The `Client` constructor accepts callback functions to handle these server requests:
MCP allows servers to interact with clients in order to provide additional capabilities. The `Client` constructor accepts additional configuration to handle these server requests.
#### Roots
* **`roots: RootsList | RootsHandler | None`**: Provides the server with a list of root directories the client grants access to. This can be a static list or a function that dynamically determines roots.
```python
from pathlib import Path
from fastmcp.client.roots import RootsHandler, RootsList
from mcp.shared.context import RequestContext # For type hint
# Option 1: Static list
static_roots: RootsList = [str(Path.home() / "Documents")]
# Option 2: Dynamic function
def dynamic_roots_handler(context: RequestContext) -> RootsList:
# Logic to determine accessible roots based on context
print(f"Server requested roots (Request ID: {context.request_id})")
return [str(Path.home() / "Downloads")]
client_with_roots = Client(
"my_server.py",
roots=dynamic_roots_handler # or roots=static_roots
)
# Tell the server the roots might have changed (if needed)
# async with client_with_roots:
# await client_with_roots.send_roots_list_changed()
```
See `fastmcp.client.roots` for helpers.
#### LLM Sampling
* **`sampling_handler: SamplingHandler | None`**: Handles `sampling/createMessage` requests from the server. This callback receives messages from the server and should return an LLM completion.
```python
from fastmcp.client.sampling import SamplingHandler, MessageResult
from mcp.types import SamplingMessage, SamplingParams, TextContent
from mcp.shared.context import RequestContext # For type hint
MCP Servers can request LLM completions from clients. The client can provide a `sampling_handler` to handle these requests. The sampling handler receives a list of messages and other parameters from the server, and should return a string completion.
async def my_llm_handler(
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext
) -> str | MessageResult:
print(f"Server requested sampling (Request ID: {context.request_id})")
# In a real scenario, call your LLM API here
last_user_message = next((m for m in reversed(messages) if m.role == 'user'), None)
prompt = last_user_message.content.text if last_user_message and isinstance(last_user_message.content, TextContent) else "Default prompt"
The following example uses the `marvin` library to generate a completion:
# Simulate LLM response
response_text = f"LLM processed: {prompt[:50]}..."
# Return simple string (becomes TextContent) or a MessageResult object
return response_text
```python {8-17, 21}
import marvin
from fastmcp import Client
from fastmcp.client.sampling import (
SamplingMessage,
SamplingParams,
RequestContext,
)
client_with_sampling = Client(
"my_server.py",
sampling_handler=my_llm_handler
async def sampling_handler(
messages: list[SamplingMessage],
params: SamplingParams,
context: RequestContext
) -> str:
return await marvin.say_async(
message=[m.content.text for m in messages],
instructions=params.systemPrompt,
)
```
See `fastmcp.client.sampling` for helpers.
client = Client(
...,
sampling_handler=sampling_handler,
)
```
#### Logging
* **`log_handler: LoggingFnT | None`**: Receives log messages sent from the server (`ctx.info`, `ctx.error`, etc.).
```python
from mcp.client.session import LoggingFnT, LogLevel
MCP servers can emit logs to clients. The client can set a logging callback to receive these logs.
def my_log_handler(level: LogLevel, message: str, logger_name: str | None):
print(f"[Server Log - {level.upper()}] {logger_name or 'default'}: {message}")
```python {4-5, 9}
from fastmcp import Client
from fastmcp.client.logging import LogHandler, LogMessage
client_with_logging = Client(
"my_server.py",
log_handler=my_log_handler
)
```
async def my_log_handler(params: LogMessage):
print(f"[Server Log - {params.level.upper()}] {params.logger or 'default'}: {params.data}")
client_with_logging = Client(
...,
log_handler=my_log_handler,
)
```
#### Roots
Roots are a way for clients to inform servers about the resources they have access to or certain boundaries on their access. The server can use this information to adjust behavior or provide more accurate responses.
Servers can request roots from clients, and clients can notify servers when their roots change.
To set the roots when creating a client, users can either provide a list of roots (which can be a list of strings) or an async function that returns a list of roots.
<CodeGroup>
```python Static Roots {5}
from fastmcp import Client
client = Client(
...,
roots=["/path/to/root1", "/path/to/root2"],
)
```
```python Dynamic Roots Callback {4-6, 10}
from fastmcp import Client
from fastmcp.client.roots import RequestContext
async def roots_callback(context: RequestContext) -> list[str]:
print(f"Server requested roots (Request ID: {context.request_id})")
return ["/path/to/root1", "/path/to/root2"]
client = Client(
...,
roots=roots_callback,
)
```
</CodeGroup>
### Utility Methods
* **`ping()`**: Sends a ping request to the server to verify connectivity.

18
src/fastmcp/client/client.py
View file

@ -5,12 +5,9 @@ from typing import Any, Literal, cast, overload
import mcp.types
from mcp import ClientSession
from mcp.client.session import (
LoggingFnT,
MessageHandlerFnT,
)
from pydantic import AnyUrl
from fastmcp.client.logging import LogHandler, MessageHandler
from fastmcp.client.roots import (
RootsHandler,
RootsList,
@ -22,7 +19,14 @@ from fastmcp.server import FastMCP
from .transports import ClientTransport, SessionKwargs, infer_transport
__all__ = ["Client", "RootsHandler", "RootsList"]
__all__ = [
"Client",
"RootsHandler",
"RootsList",
"LogHandler",
"MessageHandler",
"SamplingHandler",
]
class Client:
@ -39,8 +43,8 @@ class Client:
# Common args
roots: RootsList | RootsHandler | None = None,
sampling_handler: SamplingHandler | None = None,
log_handler: LoggingFnT | None = None,
message_handler: MessageHandlerFnT | None = None,
log_handler: LogHandler | None = None,
message_handler: MessageHandler | None = None,
read_timeout_seconds: datetime.timedelta | None = None,
):
self.transport = infer_transport(transport)

13
src/fastmcp/client/logging.py Normal file
View file

@ -0,0 +1,13 @@
from typing import TypeAlias
from mcp.client.session import (
LoggingFnT,
MessageHandlerFnT,
)
from mcp.types import LoggingMessageNotificationParams
LogMessage: TypeAlias = LoggingMessageNotificationParams
LogHandler: TypeAlias = LoggingFnT
MessageHandler: TypeAlias = MessageHandlerFnT
__all__ = ["LogMessage", "LogHandler", "MessageHandler"]

2
src/fastmcp/client/sampling.py
View file

@ -9,6 +9,8 @@ from mcp.shared.context import LifespanContextT, RequestContext
from mcp.types import CreateMessageRequestParams as SamplingParams
from mcp.types import SamplingMessage
__all__ = ["SamplingMessage", "SamplingParams", "MessageResult", "SamplingHandler"]
class MessageResult(CreateMessageResult):
role: mcp.types.Role = "assistant"

28
src/fastmcp/server/context.py
View file

@ -1,7 +1,8 @@
from __future__ import annotations as _annotations
from typing import Any, Generic, Literal
from typing import Any, Generic
from mcp import LoggingLevel
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT, RequestContext
@ -122,19 +123,20 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
async def log(
self,
level: Literal["debug", "info", "warning", "error"],
message: str,
*,
level: LoggingLevel | None = None,
logger_name: str | None = None,
) -> None:
"""Send a log message to the client.
Args:
level: Log level (debug, info, warning, error)
message: Log message
level: Optional log level. One of "debug", "info", "notice", "warning", "error", "critical",
"alert", or "emergency". Default is "info".
logger_name: Optional logger name
**extra: Additional structured data to include
"""
if level is None:
level = "info"
await self.request_context.session.send_log_message(
level=level, data=message, logger=logger_name
)
@ -159,21 +161,21 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
return self.request_context.session
# Convenience methods for common log levels
async def debug(self, message: str, **extra: Any) -> None:
async def debug(self, message: str, logger_name: str | None = None) -> None:
"""Send a debug log message."""
await self.log("debug", message, **extra)
await self.log(level="debug", message=message, logger_name=logger_name)
async def info(self, message: str, **extra: Any) -> None:
async def info(self, message: str, logger_name: str | None = None) -> None:
"""Send an info log message."""
await self.log("info", message, **extra)
await self.log(level="info", message=message, logger_name=logger_name)
async def warning(self, message: str, **extra: Any) -> None:
async def warning(self, message: str, logger_name: str | None = None) -> None:
"""Send a warning log message."""
await self.log("warning", message, **extra)
await self.log(level="warning", message=message, logger_name=logger_name)
async def error(self, message: str, **extra: Any) -> None:
async def error(self, message: str, logger_name: str | None = None) -> None:
"""Send an error log message."""
await self.log("error", message, **extra)
await self.log(level="error", message=message, logger_name=logger_name)
async def list_roots(self) -> list[Root]:
"""List the roots available to the server, as indicated by the client."""

60
tests/client/test_logs.py Normal file
View file

@ -0,0 +1,60 @@
import pytest
from mcp import LoggingLevel
from fastmcp import Client, Context, FastMCP
from fastmcp.client.logging import LogMessage
class LogHandler:
def __init__(self):
self.logs: list[LogMessage] = []
async def handle_log(self, params: LogMessage) -> None:
self.logs.append(params)
@pytest.fixture
def fastmcp_server():
mcp = FastMCP()
@mcp.tool()
async def log(context: Context) -> None:
await context.info(message="hello?")
@mcp.tool()
async def echo_log(
message: str,
context: Context,
level: LoggingLevel | None = None,
logger: str | None = None,
) -> None:
await context.log(message=message, level=level)
return mcp
class TestClientLogs:
async def test_log(self, fastmcp_server: FastMCP):
log_handler = LogHandler()
async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
await client.call_tool("log", {})
assert len(log_handler.logs) == 1
assert log_handler.logs[0].data == "hello?"
assert log_handler.logs[0].level == "info"
async def test_echo_log(self, fastmcp_server: FastMCP):
log_handler = LogHandler()
async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
await client.call_tool("echo_log", {"message": "this is a log"})
assert len(log_handler.logs) == 1
await client.call_tool(
"echo_log", {"message": "this is a warning log", "level": "warning"}
)
assert len(log_handler.logs) == 2
assert log_handler.logs[0].data == "this is a log"
assert log_handler.logs[0].level == "info"
assert log_handler.logs[1].data == "this is a warning log"
assert log_handler.logs[1].level == "warning"

43
tests/server/test_server_interactions.py
View file

@ -689,10 +689,10 @@ class TestToolContextInjection:
"""Test that context parameters are properly detected."""
mcp = FastMCP()
@mcp.tool()
def tool_with_context(x: int, ctx: Context) -> str:
return f"Request {ctx.request_id}: {x}"
mcp.add_tool(tool_with_context)
async with Client(mcp) as client:
tools = await client.list_tools()
assert len(tools) == 1
@ -719,11 +719,11 @@ class TestToolContextInjection:
"""Test that context works in async functions."""
mcp = FastMCP()
@mcp.tool()
async def async_tool(x: int, ctx: Context) -> str:
assert ctx.request_id is not None
return f"Async request {ctx.request_id}: {x}"
mcp.add_tool(async_tool)
async with Client(mcp) as client:
result = await client.call_tool("async_tool", {"x": 42})
assert len(result) == 1
@ -732,51 +732,14 @@ class TestToolContextInjection:
assert "Async request" in content.text
assert "42" in content.text
async def test_context_logging(self):
from unittest.mock import patch
import mcp.server.session
"""Test that context logging methods work."""
mcp = FastMCP()
async def logging_tool(msg: str, ctx: Context) -> str:
await ctx.debug("Debug message")
await ctx.info("Info message")
await ctx.warning("Warning message")
await ctx.error("Error message")
return f"Logged messages for {msg}"
mcp.add_tool(logging_tool)
with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
async with Client(mcp) as client:
result = await client.call_tool("logging_tool", {"msg": "test"})
assert len(result) == 1
content = result[0]
assert isinstance(content, TextContent)
assert "Logged messages for test" in content.text
assert mock_log.call_count == 4
mock_log.assert_any_call(
level="debug", data="Debug message", logger=None
)
mock_log.assert_any_call(level="info", data="Info message", logger=None)
mock_log.assert_any_call(
level="warning", data="Warning message", logger=None
)
mock_log.assert_any_call(
level="error", data="Error message", logger=None
)
async def test_optional_context(self):
"""Test that context is optional."""
mcp = FastMCP()
@mcp.tool()
def no_context(x: int) -> int:
return x * 2
mcp.add_tool(no_context)
async with Client(mcp) as client:
result = await client.call_tool("no_context", {"x": 21})
assert len(result) == 1