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

add client docs

Browse source
This commit is contained in:
Jeremiah Lowin 2025-04-13 12:42:40 -04:00
commit 3707be0700
4 changed files with 510 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

252
docs/clients/overview.mdx Normal file
View file

@ -0,0 +1,252 @@
---
title: Client Overview
sidebarTitle: Overview
description: Learn how to use the FastMCP Client to interact with MCP servers.
icon: user-robot
---
The `fastmcp.Client` provides a high-level, asynchronous interface for interacting with any Model Context Protocol (MCP) server, whether it's built with FastMCP or another implementation. It simplifies communication by handling protocol details and connection management.
## FastMCP Client
The FastMCP Client architecture separates the protocol logic (`Client`) from the connection mechanism (`Transport`).
- **`Client`**: Handles sending MCP requests (like `tools/call`, `resources/read`), receiving responses, and managing callbacks.
- **`Transport`**: Responsible for establishing and maintaining the connection to the server (e.g., via WebSockets, SSE, Stdio, or in-memory).
### Transports
Clients must be initialized with a `transport`. You can either provide an already instantiated transport object, or provide a transport source and let FastMCP attempt to infer the correct transport to use.
The following inference rules are used to determine the appropriate `ClientTransport` based on the input type:
1. **`ClientTransport` Instance**: If you provide an already instantiated transport object, it's used directly.
2. **`FastMCP` Instance**: Creates a `FastMCPTransport` for efficient in-memory communication (ideal for testing).
3. **`Path` or `str` pointing to an existing file**:
* If it ends with `.py`: Creates a `PythonStdioTransport` to run the script using `python`.
* If it ends with `.js`: Creates a `NodeStdioTransport` to run the script using `node`.
4. **`AnyUrl` or `str` pointing to a URL**:
* If it starts with `http://` or `https://`: Creates an `SSETransport`.
* If it starts with `ws://` or `wss://`: Creates a `WSTransport`.
5. **Other**: Raises a `ValueError` if the type cannot be inferred.
```python
import asyncio
from fastmcp import Client, FastMCP
# Example transports (more details in Transports page)
server_instance = FastMCP(name="TestServer") # In-memory server
sse_url = "http://localhost:8000/sse" # SSE server URL
ws_url = "ws://localhost:9000" # WebSocket server URL
server_script = "my_mcp_server.py" # Path to a Python server file
# Client automatically infers the transport type
client_in_memory = Client(server_instance)
client_sse = Client(sse_url)
client_ws = Client(ws_url)
client_stdio = Client(server_script)
print(client_in_memory.transport)
print(client_sse.transport)
print(client_ws.transport)
print(client_stdio.transport)
# Expected Output (types may vary slightly based on environment):
# <FastMCP(server='TestServer')>
# <SSE(url='http://localhost:8000/sse')>
# <WebSocket(url='ws://localhost:9000')>
# <PythonStdioTransport(command='python', args=['/path/to/your/my_mcp_server.py'])>
```
<Tip>
For more control over connection details (like headers for SSE, environment variables for Stdio), you can instantiate the specific `ClientTransport` class yourself and pass it to the `Client`. See the [Transports](/clients/transports) page for details.
</Tip>
## Client Usage
### Connection Lifecycle
The client operates asynchronously and must be used within an `async with` block. This context manager handles establishing the connection, initializing the MCP session, and cleaning up resources upon exit.
```python
import asyncio
from fastmcp import Client
client = Client("my_mcp_server.py") # Assumes my_mcp_server.py exists
async def main():
# Connection is established here
async with client:
print(f"Client connected: {client.is_connected()}")
# Make MCP calls within the context
tools = await client.list_tools()
print(f"Available tools: {tools}")
if any(tool.name == "greet" for tool in tools):
result = await client.call_tool("greet", {"name": "World"})
print(f"Greet result: {result}")
# Connection is closed automatically here
print(f"Client connected: {client.is_connected()}")
if __name__ == "__main__":
asyncio.run(main())
```
You can make multiple calls to the server within the same `async with` block using the established session.
### Client Methods
The `Client` provides methods corresponding to standard MCP requests:
#### Tool Operations
* **`list_tools()`**: Retrieves a list of tools available on the server.
```python
tools = await client.list_tools()
# tools -> list[mcp.types.Tool]
```
* **`call_tool(name: str, arguments: dict[str, Any] | None = None)`**: Executes a tool on the server.
```python
result = await client.call_tool("add", {"a": 5, "b": 3})
# result -> list[mcp.types.TextContent | mcp.types.ImageContent | ...]
print(result[0].text) # Assuming TextContent, e.g., '8'
```
* Arguments are passed as a dictionary. FastMCP servers automatically handle JSON string parsing for complex types if needed.
* Returns a list of content objects (usually `TextContent` or `ImageContent`).
#### Resource Operations
* **`list_resources()`**: Retrieves a list of static resources.
```python
resources = await client.list_resources()
# resources -> list[mcp.types.Resource]
```
* **`list_resource_templates()`**: Retrieves a list of resource templates.
```python
templates = await client.list_resource_templates()
# templates -> list[mcp.types.ResourceTemplate]
```
* **`read_resource(uri: str | AnyUrl)`**: Reads the content of a resource or a resolved template.
```python
# Read a static resource
readme_content = await client.read_resource("file:///path/to/README.md")
# readme_content -> list[mcp.types.TextResourceContents | mcp.types.BlobResourceContents]
print(readme_content[0].text) # Assuming text
# Read a resource generated from a template
weather_content = await client.read_resource("data://weather/london")
print(weather_content[0].text) # Assuming text JSON
```
#### Prompt Operations
* **`list_prompts()`**: Retrieves available prompt templates.
* **`get_prompt(name: str, arguments: dict[str, Any] | None = None)`**: Retrieves a rendered prompt message list.
### Callbacks
MCP allows servers to make requests *back* to the client for certain capabilities. The `Client` constructor accepts callback functions 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
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"
# Simulate LLM response
response_text = f"LLM processed: {prompt[:50]}..."
# Return simple string (becomes TextContent) or a MessageResult object
return response_text
client_with_sampling = Client(
"my_server.py",
sampling_handler=my_llm_handler
)
```
See `fastmcp.client.sampling` for helpers.
#### 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
def my_log_handler(level: LogLevel, message: str, logger_name: str | None):
print(f"[Server Log - {level.upper()}] {logger_name or 'default'}: {message}")
client_with_logging = Client(
"my_server.py",
log_handler=my_log_handler
)
```
### Error Handling
When a `call_tool` request results in an error on the server (e.g., the tool function raised an exception), the `client.call_tool()` method will raise a `fastmcp.client.ClientError`.
```python
async def safe_call_tool():
async with client:
try:
# Assume 'divide' tool exists and might raise ZeroDivisionError
result = await client.call_tool("divide", {"a": 10, "b": 0})
print(f"Result: {result}")
except ClientError as e:
print(f"Tool call failed: {e}")
except ConnectionError as e:
print(f"Connection failed: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
# Example Output if division by zero occurs:
# Tool call failed: Division by zero is not allowed.
```
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_tools(..., return_raw_result=True)` to get the raw result object and handle errors yourself by checking its `isError` attribute.
</Tip>

241
docs/clients/transports.mdx Normal file
View file

@ -0,0 +1,241 @@
---
title: Client Transports
sidebarTitle: Transports
description: Understand the different ways FastMCP Clients can connect to servers.
icon: link
---
The FastMCP `Client` relies on a `ClientTransport` object to handle the specifics of connecting to and communicating with an MCP server. FastMCP provides several built-in transport implementations for common connection methods.
While the `Client` often infers the correct transport automatically (see [Client Overview](/clients/overview#transport-inference)), you can also instantiate transports explicitly for more control.
## Stdio Transports
These transports manage an MCP server running as a subprocess, communicating with it via standard input (stdin) and standard output (stdout). This is the standard mechanism used by clients like Claude Desktop.
### Python Stdio
* **Class:** `fastmcp.client.transports.PythonStdioTransport`
* **Inferred From:** Paths to `.py` files.
* **Use Case:** Running a Python-based MCP server script (like one using FastMCP or the base `mcp` library) in a subprocess.
This is the most common way to interact with local FastMCP servers during development or when integrating with tools that expect to launch a server script.
```python
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport
server_script = "my_mcp_server.py" # Assumes this file exists and runs mcp.run()
# Option 1: Inferred transport
client_inferred = Client(server_script)
# Option 2: Explicit transport (e.g., to use a specific python executable or add args)
transport_explicit = PythonStdioTransport(
script_path=server_script,
python_cmd="/usr/bin/python3.11", # Specify python version
# args=["--some-server-arg"], # Pass args to the script
# env={"MY_VAR": "value"}, # Set environment variables
# cwd="/path/to/run/in" # Set working directory
)
client_explicit = Client(transport_explicit)
async def use_stdio_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via Python Stdio, found tools: {tools}")
# asyncio.run(use_stdio_client(client_inferred))
# asyncio.run(use_stdio_client(client_explicit))
```
<Warning>
The server script (`my_mcp_server.py` in the example) *must* include logic to start the MCP server and listen on stdio, typically via `mcp.run()` or `fastmcp.server.run()`. The `Client` only launches the script; it doesn't inject the server logic.
</Warning>
### Node.js Stdio
* **Class:** `fastmcp.client.transports.NodeStdioTransport`
* **Inferred From:** Paths to `.js` files.
* **Use Case:** Running a Node.js-based MCP server script in a subprocess.
Similar to the Python transport, but for JavaScript servers.
```python
from fastmcp import Client
from fastmcp.client.transports import NodeStdioTransport
node_server_script = "my_mcp_server.js" # Assumes this JS file starts an MCP server on stdio
# Option 1: Inferred transport
client_inferred = Client(node_server_script)
# Option 2: Explicit transport
transport_explicit = NodeStdioTransport(
script_path=node_server_script,
node_cmd="node" # Or specify path to Node executable
)
client_explicit = Client(transport_explicit)
# Usage is the same as other clients
# async with client_explicit:
# tools = await client_explicit.list_tools()
```
### UVX Stdio (Experimental)
* **Class:** `fastmcp.client.transports.UvxStdioTransport`
* **Inferred From:** Not automatically inferred. Must be instantiated explicitly.
* **Use Case:** Running an MCP server packaged as a Python tool using [`uvx`](https://docs.astral.sh/uv/reference/cli/#uvx) (part of the `uv` toolchain). This allows running tools without explicitly installing them into the current environment.
This is useful for executing MCP servers distributed as command-line tools or packages.
```python
from fastmcp.client.transports import UvxStdioTransport
# Example: Run a hypothetical 'cloud-analyzer-mcp' tool via uvx
# Assume this tool, when run, starts an MCP server on stdio
transport = UvxStdioTransport(
tool_name="cloud-analyzer-mcp",
# from_package="cloud-analyzer-cli", # Optionally specify package if tool name differs
# with_packages=["boto3", "requests"], # Add dependencies if needed
# tool_args=["--config", "prod.yaml"] # Pass args to the tool itself
)
client = Client(transport)
# async with client:
# analysis = await client.call_tool("analyze_bucket", {"name": "my-data"})
```
### NPX Stdio (Experimental)
* **Class:** `fastmcp.client.transports.NpxStdioTransport`
* **Inferred From:** Not automatically inferred. Must be instantiated explicitly.
* **Use Case:** Running an MCP server packaged as an NPM package using `npx`.
Similar to `UvxStdioTransport`, but for the Node.js ecosystem.
```python
from fastmcp.client.transports import NpxStdioTransport
# Example: Run a hypothetical 'npm-mcp-server-package' via npx
transport = NpxStdioTransport(
package="npm-mcp-server-package",
# args=["--port", "stdio"] # Args passed to the package script
)
client = Client(transport)
# async with client:
# response = await client.call_tool("get_npm_data", {})
```
## Network Transports
These transports connect to servers running over a network, typically long-running services accessible via URLs.
### SSE (Server-Sent Events)
* **Class:** `fastmcp.client.transports.SSETransport`
* **Inferred From:** `http://` or `https://` URLs
* **Use Case:** Connecting to persistent MCP servers exposed over HTTP/S, often using FastMCP's `mcp.run(transport="sse")` mode.
SSE is a simple, unidirectional protocol where the server pushes messages to the client over a standard HTTP connection.
```python
from fastmcp import Client
from fastmcp.client.transports import SSETransport
sse_url = "http://localhost:8000/sse"
# Option 1: Inferred transport
client_inferred = Client(sse_url)
# Option 2: Explicit transport (e.g., to add custom headers)
headers = {"Authorization": "Bearer mytoken"}
transport_explicit = SSETransport(url=sse_url, headers=headers)
client_explicit = Client(transport_explicit)
async def use_sse_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via SSE, found tools: {tools}")
# asyncio.run(use_sse_client(client_inferred))
# asyncio.run(use_sse_client(client_explicit))
```
### WebSocket
* **Class:** `fastmcp.client.transports.WSTransport`
* **Inferred From:** `ws://` or `wss://` URLs
* **Use Case:** Connecting to MCP servers using the WebSocket protocol for bidirectional communication.
WebSockets provide a persistent, full-duplex connection between client and server.
```python
from fastmcp import Client
from fastmcp.client.transports import WSTransport
ws_url = "ws://localhost:9000"
# Option 1: Inferred transport
client_inferred = Client(ws_url)
# Option 2: Explicit transport
transport_explicit = WSTransport(url=ws_url)
client_explicit = Client(transport_explicit)
async def use_ws_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via WebSocket, found tools: {tools}")
# asyncio.run(use_ws_client(client_inferred))
# asyncio.run(use_ws_client(client_explicit))
```
## In-Memory Transports
### FastMCP Transport
* **Class:** `fastmcp.client.transports.FastMCPTransport`
* **Inferred From:** An instance of `fastmcp.server.FastMCP`.
* **Use Case:** Connecting directly to a `FastMCP` server instance running in the *same Python process*.
This is extremely useful for:
* **Testing:** Writing unit or integration tests for your FastMCP server without needing subprocesses or network connections.
* **Embedding:** Using an MCP server as a component within a larger application.
```python
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport
# 1. Create your FastMCP server instance
server = FastMCP(name="InMemoryServer")
@server.tool()
def ping(): return "pong"
# 2. Create a client pointing directly to the server instance
# Option A: Inferred
client_inferred = Client(server)
# Option B: Explicit
transport_explicit = FastMCPTransport(mcp=server)
client_explicit = Client(transport_explicit)
# 3. Use the client (no subprocess or network involved)
async def test_in_memory():
async with client_inferred: # Or client_explicit
result = await client_inferred.call_tool("ping")
print(f"In-memory call result: {result[0].text}") # Output: pong
# asyncio.run(test_in_memory())
```
Communication happens through efficient in-memory queues, making it very fast.
## Choosing a Transport
* **Local Development/Testing:** Use `PythonStdioTransport` (inferred from `.py` files) or `FastMCPTransport` (for same-process testing).
* **Connecting to Remote/Persistent Servers:** Use `SSETransport` (for `http/s`) or `WSTransport` (for `ws/s`).
* **Running Packaged Tools:** Use `UvxStdioTransport` (Python/uv) or `NpxStdioTransport` (Node/npm) if you need to run MCP servers without local installation.
* **Integrating with Claude Desktop (or similar):** These tools typically expect to run a Python script, so your server should be runnable via `python your_server.py`, making `PythonStdioTransport` the relevant mechanism on the client side.

9
docs/docs.json
View file

@ -15,7 +15,9 @@
"description": "The fast, Pythonic way to build MCP servers.",
"footer": {
"socials": {
"github": "https://github.com/jlowin/fastmcp"
"bluesky": "https://bsky.app/profile/jlowin.dev",
"github": "https://github.com/jlowin/fastmcp",
"x": "https://x.com/jlowin"
}
},
"name": "FastMCP",
@ -47,7 +49,10 @@
},
{
"group": "Clients",
"pages": []
"pages": [
"clients/overview",
"clients/transports"
]
},
{
"group": "Deployment",

20
docs/style.css
View file

@ -1,13 +1,13 @@
/* Target inline code elements with higher specificity */
p code,
table code,
li code,
h1 code,
h2 code,
h3 code,
h4 code,
h5 code,
h6 code {
/* Target only inline code elements, not code blocks */
p code:not(pre code),
table code:not(pre code),
li code:not(pre code),
h1 code:not(pre code),
h2 code:not(pre code),
h3 code:not(pre code),
h4 code:not(pre code),
h5 code:not(pre code),
h6 code:not(pre code) {
color: #f72585 !important;
background-color: #ea54551a !important;
}