mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-27 15:51:33 +02:00
320 lines
No EOL
10 KiB
Text
320 lines
No EOL
10 KiB
Text
---
|
|
title: Client Transports
|
|
sidebarTitle: Transports
|
|
description: Understand the different ways FastMCP Clients can connect to servers.
|
|
icon: link
|
|
---
|
|
|
|
import { VersionBadge } from "/snippets/version-badge.mdx"
|
|
|
|
<VersionBadge version="2.0.0" />
|
|
|
|
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/client#transport-inference)), you can also instantiate transports explicitly for more control.
|
|
|
|
<Tip>
|
|
Clients are lightweight objects, so don't hesitate to create new ones as needed. However, be mindful of the context management - each time you open a client context (`async with client:`), a new connection or process starts. For best performance, keep client contexts open while performing multiple operations rather than repeatedly opening and closing them.
|
|
</Tip>
|
|
|
|
## Choosing a Transport
|
|
|
|
Choose the transport that best fits your use case:
|
|
|
|
- **Connecting to Remote/Persistent Servers:** Use `StreamableHttpTransport` (recommended, default for HTTP URLs) or `SSETransport` (legacy option) for web-based deployments.
|
|
|
|
- **Local Development/Testing:** Use `FastMCPTransport` for in-memory, same-process testing of your FastMCP servers.
|
|
|
|
- **Running Local Servers:** Use `UvxStdioTransport` (Python/uv) or `NpxStdioTransport` (Node/npm) if you need to run MCP servers as packaged tools.
|
|
|
|
## Network Transports
|
|
|
|
These transports connect to servers running over a network, typically long-running services accessible via URLs.
|
|
|
|
### Streamable HTTP
|
|
|
|
<VersionBadge version="2.3.0" />
|
|
|
|
Streamable HTTP is the recommended transport for web-based deployments, providing efficient bidirectional communication over HTTP.
|
|
|
|
#### Overview
|
|
|
|
- **Class:** `fastmcp.client.transports.StreamableHttpTransport`
|
|
- **Inferred From:** URLs starting with `http://` or `https://` (default for HTTP URLs since v2.3.0) that do not contain `/sse/` in the path
|
|
- **Server Compatibility:** Works with FastMCP servers running in `streamable-http` mode
|
|
|
|
#### Basic Usage
|
|
|
|
The simplest way to use Streamable HTTP is to let the transport be inferred from a URL:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
import asyncio
|
|
|
|
# The Client automatically uses StreamableHttpTransport for HTTP URLs
|
|
client = Client("https://example.com/mcp")
|
|
|
|
async def main():
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
print(f"Available tools: {tools}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
You can also explicitly instantiate the transport:
|
|
|
|
```python
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
|
|
|
transport = StreamableHttpTransport(url="https://example.com/mcp")
|
|
client = Client(transport)
|
|
```
|
|
|
|
#### Authentication with Headers
|
|
|
|
For servers requiring authentication:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
|
|
|
# Create transport with authentication headers
|
|
transport = StreamableHttpTransport(
|
|
url="https://example.com/mcp",
|
|
headers={"Authorization": "Bearer your-token-here"}
|
|
)
|
|
|
|
client = Client(transport)
|
|
```
|
|
|
|
### SSE (Server-Sent Events)
|
|
|
|
<VersionBadge version="2.0.0" />
|
|
|
|
Server-Sent Events (SSE) is a transport that allows servers to push data to clients over HTTP connections. While still supported, Streamable HTTP is now the recommended transport for new web-based deployments.
|
|
|
|
#### Overview
|
|
|
|
- **Class:** `fastmcp.client.transports.SSETransport`
|
|
- **Inferred From:** HTTP URLs containing `/sse/` in the path
|
|
- **Server Compatibility:** Works with FastMCP servers running in `sse` mode
|
|
|
|
#### Basic Usage
|
|
|
|
The simplest way to use SSE is to let the transport be inferred from a URL with `/sse/` in the path:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
import asyncio
|
|
|
|
# The Client automatically uses SSETransport for URLs containing /sse/ in the path
|
|
client = Client("https://example.com/sse")
|
|
|
|
async def main():
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
print(f"Available tools: {tools}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
You can also explicitly instantiate the transport for URLs that do not contain `/sse/` in the path or for more control:
|
|
|
|
```python
|
|
from fastmcp.client.transports import SSETransport
|
|
|
|
transport = SSETransport(url="https://example.com/sse")
|
|
client = Client(transport)
|
|
```
|
|
|
|
#### Authentication with Headers
|
|
|
|
SSE transport also supports custom headers for authentication:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
from fastmcp.client.transports import SSETransport
|
|
|
|
# Create SSE transport with authentication headers
|
|
transport = SSETransport(
|
|
url="https://example.com/sse",
|
|
headers={"Authorization": "Bearer your-token-here"}
|
|
)
|
|
|
|
client = Client(transport)
|
|
```
|
|
|
|
#### When to Use SSE vs. Streamable HTTP
|
|
|
|
- **Use Streamable HTTP when:**
|
|
- Setting up new deployments (recommended default)
|
|
- You need bidirectional streaming
|
|
- You're connecting to FastMCP servers running in `streamable-http` mode
|
|
|
|
- **Use SSE when:**
|
|
- Connecting to legacy FastMCP servers running in `sse` mode
|
|
- Working with infrastructure optimized for Server-Sent Events
|
|
|
|
## Local 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 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" # Path to your server script
|
|
|
|
# Option 1: Inferred transport
|
|
client = Client(server_script)
|
|
|
|
# Option 2: Explicit transport with custom configuration
|
|
transport = PythonStdioTransport(
|
|
script_path=server_script,
|
|
python_cmd="/usr/bin/python3.11", # Optional: specify Python interpreter
|
|
# args=["--some-server-arg"], # Optional: pass arguments to the script
|
|
# env={"MY_VAR": "value"}, # Optional: set environment variables
|
|
)
|
|
client = Client(transport)
|
|
|
|
async def main():
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
print(f"Connected via Python Stdio, found tools: {tools}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
<Warning>
|
|
The server script 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" # Path to your Node.js server script
|
|
|
|
# Option 1: Inferred transport
|
|
client = Client(node_server_script)
|
|
|
|
# Option 2: Explicit transport
|
|
transport = NodeStdioTransport(
|
|
script_path=node_server_script,
|
|
node_cmd="node" # Optional: specify path to Node executable
|
|
)
|
|
client = Client(transport)
|
|
|
|
async def main():
|
|
async with client:
|
|
tools = await client.list_tools()
|
|
print(f"Connected via Node.js Stdio, found tools: {tools}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
### UVX Stdio (Experimental)
|
|
|
|
- **Class:** `fastmcp.client.transports.UvxStdioTransport`
|
|
- **Inferred From:** Not automatically inferred
|
|
- **Use Case:** Running an MCP server packaged as a Python tool using [`uvx`](https://docs.astral.sh/uv/reference/cli/#uvx)
|
|
|
|
This is useful for executing MCP servers distributed as command-line tools or packages without installing them into your environment.
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
from fastmcp.client.transports import UvxStdioTransport
|
|
|
|
# Run a hypothetical 'cloud-analyzer-mcp' tool via uvx
|
|
transport = UvxStdioTransport(
|
|
tool_name="cloud-analyzer-mcp",
|
|
# from_package="cloud-analyzer-cli", # Optional: specify package if tool name differs
|
|
# with_packages=["boto3", "requests"] # Optional: add dependencies
|
|
)
|
|
client = Client(transport)
|
|
|
|
async def main():
|
|
async with client:
|
|
result = await client.call_tool("analyze_bucket", {"name": "my-data"})
|
|
print(f"Analysis result: {result}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
### NPX Stdio (Experimental)
|
|
|
|
- **Class:** `fastmcp.client.transports.NpxStdioTransport`
|
|
- **Inferred From:** Not automatically inferred
|
|
- **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 import Client
|
|
from fastmcp.client.transports import NpxStdioTransport
|
|
|
|
# Run an MCP server from an NPM package
|
|
transport = NpxStdioTransport(
|
|
package="mcp-server-package",
|
|
# args=["--port", "stdio"] # Optional: pass arguments to the package
|
|
)
|
|
client = Client(transport)
|
|
|
|
async def main():
|
|
async with client:
|
|
result = await client.call_tool("get_npm_data", {})
|
|
print(f"Result: {result}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
## 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 in the same Python process
|
|
|
|
This is extremely useful for testing your FastMCP servers.
|
|
|
|
```python
|
|
from fastmcp import FastMCP, Client
|
|
import asyncio
|
|
|
|
# 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
|
|
client = Client(server) # Transport is automatically inferred
|
|
|
|
async def main():
|
|
async with client:
|
|
result = await client.call_tool("ping")
|
|
print(f"In-memory call result: {result}")
|
|
|
|
asyncio.run(main())
|
|
```
|
|
|
|
Communication happens through efficient in-memory queues, making it very fast and ideal for unit testing. |