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

Update docs and test

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-20 13:06:48 -04:00
commit 18ae625fef
19 changed files with 51 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/deployment/asgi.mdx
View file

@ -48,7 +48,7 @@ Both approaches return a Starlette application that can be integrated with other
The returned app stores the `FastMCP` instance on `app.state.fastmcp_server`, so you
can access it from custom middleware or routes via `request.app.state.fastmcp_server`.
The MCP server's endpoint is mounted at the root path `/mcp` for Streamable HTTP transport, and `/sse` for SSE transport, though you can change these paths by passing a `path` argument to the `http_app()` method:
The MCP server's endpoint is mounted at the root path `/mcp/` for Streamable HTTP transport, and `/sse/` for SSE transport, though you can change these paths by passing a `path` argument to the `http_app()` method:
```python
# For Streamable HTTP transport
@ -137,7 +137,7 @@ app = Starlette(
)
```
The MCP endpoint will be available at `/mcp-server/mcp` of the resulting Starlette app.
The MCP endpoint will be available at `/mcp-server/mcp/` of the resulting Starlette app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the resulting Starlette app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.
@ -167,7 +167,7 @@ app = Starlette(
)
```
In this setup, the MCP server is accessible at the `/outer/inner/mcp` path of the resulting Starlette app.
In this setup, the MCP server is accessible at the `/outer/inner/mcp/` path of the resulting Starlette app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the *outer* Starlette app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.
@ -194,7 +194,7 @@ app = FastAPI(lifespan=mcp_app.lifespan)
app.mount("/mcp-server", mcp_app)
```
The MCP endpoint will be available at `/mcp-server/mcp` of the resulting FastAPI app.
The MCP endpoint will be available at `/mcp-server/mcp/` of the resulting FastAPI app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the resulting FastAPI app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.

8
docs/deployment/running-server.mdx
View file

@ -105,7 +105,7 @@ When using Stdio transport, you will typically *not* run the server yourself as
Streamable HTTP is a modern, efficient transport for exposing your MCP server via HTTP. It is the recommended transport for web-based deployments.
To run a server using Streamable HTTP, you can use the `run()` method with the `transport` argument set to `"streamable-http"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and path (`/mcp`).
To run a server using Streamable HTTP, you can use the `run()` method with the `transport` argument set to `"streamable-http"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and path (`/mcp/`).
<CodeGroup>
```python {6} server.py
from fastmcp import FastMCP
@ -120,7 +120,7 @@ import asyncio
from fastmcp import Client
async def example():
async with Client("http://127.0.0.1:8000/mcp") as client:
async with Client("http://127.0.0.1:8000/mcp/") as client:
await client.ping()
if __name__ == "__main__":
@ -168,7 +168,7 @@ New applications should use Streamable HTTP transport instead.
Server-Sent Events (SSE) is an HTTP-based protocol for server-to-client streaming. While FastMCP still supports SSE, it is deprecated and Streamable HTTP is preferred for new projects.
To run a server using SSE, you can use the `run()` method with the `transport` argument set to `"sse"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and with default SSE path (`/sse`) and message path (`/messages/`).
To run a server using SSE, you can use the `run()` method with the `transport` argument set to `"sse"`. This will start a Uvicorn server on the default host (`127.0.0.1`), port (`8000`), and with default SSE path (`/sse/`) and message path (`/messages/`).
<CodeGroup>
```python {6} server.py
@ -186,7 +186,7 @@ from fastmcp.client.transports import SSETransport
async def example():
async with Client(
transport=SSETransport("http://127.0.0.1:8000/sse")
transport=SSETransport("http://127.0.0.1:8000/sse/")
) as client:
await client.ping()

2
docs/tutorials/rest-api.mdx
View file

@ -103,7 +103,7 @@ from fastmcp import Client
async def main():
# Connect to the MCP server we just created
async with Client("http://127.0.0.1:8000/mcp") as client:
async with Client("http://127.0.0.1:8000/mcp/") as client:
# List the tools that were automatically generated
tools = await client.list_tools()

2
src/fastmcp/client/auth/oauth.py
View file

@ -307,7 +307,7 @@ def OAuth(
Args:
mcp_url: Full URL to the MCP endpoint (e.g.,
"http://host/mcp/sse")
"http://host/mcp/sse/")
scopes: OAuth scopes to request. Can be a
space-separated string or a list of strings.
client_name: Name for this client during registration

4
src/fastmcp/utilities/mcp_config.py
View file

@ -1,5 +1,6 @@
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Annotated, Any, Literal
from urllib.parse import urlparse
@ -28,7 +29,8 @@ def infer_transport_type_from_url(
parsed_url = urlparse(url)
path = parsed_url.path
if "/sse/" in path or path.rstrip("/").endswith("/sse"):
# Match /sse followed by /, ?, &, or end of string
if re.search(r"/sse(/|\?|&|$)", path):
return "sse"
else:
return "streamable-http"

6
tests/auth/providers/test_bearer.py
View file

@ -67,7 +67,7 @@ def mcp_server_url(rsa_key_pair: RSAKeyPair) -> Generator[str]:
public_key=rsa_key_pair.public_key,
run_kwargs=dict(transport="streamable-http"),
) as url:
yield f"{url}/mcp"
yield f"{url}/mcp/"
class TestRSAKeyPair:
@ -698,7 +698,7 @@ class TestFastMCPBearerAuth:
auth_kwargs=dict(required_scopes=["read", "write"]),
run_kwargs=dict(transport="streamable-http"),
) as url:
mcp_server_url = f"{url}/mcp"
mcp_server_url = f"{url}/mcp/"
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
tools = await client.list_tools() # noqa: F841
@ -721,7 +721,7 @@ class TestFastMCPBearerAuth:
auth_kwargs=dict(required_scopes=["read", "write"]),
run_kwargs=dict(transport="streamable-http"),
) as url:
mcp_server_url = f"{url}/mcp"
mcp_server_url = f"{url}/mcp/"
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
tools = await client.list_tools()
assert tools

2
tests/auth/test_oauth_client.py
View file

@ -44,7 +44,7 @@ def run_server(host: str, port: int, **kwargs) -> None:
@pytest.fixture(scope="module")
def streamable_http_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="streamable-http") as url:
yield f"{url}/mcp"
yield f"{url}/mcp/"
@pytest.fixture()

12
tests/client/test_client.py
View file

@ -735,7 +735,8 @@ class TestInferTransport:
"http://example.com/api/sse/stream",
"https://localhost:8080/mcp/sse/endpoint",
"http://example.com/api/sse",
"https://localhost:8080/mcp/sse",
"http://example.com/api/sse/",
"https://localhost:8080/mcp/sse/",
"http://example.com/api/sse?param=value",
"https://localhost:8080/mcp/sse/?param=value",
"https://localhost:8000/mcp/sse?x=1&y=2",
@ -744,6 +745,7 @@ class TestInferTransport:
"path_with_sse_directory",
"path_with_sse_subdirectory",
"path_ending_with_sse",
"path_ending_with_sse_slash",
"path_ending_with_sse_https",
"path_with_sse_and_query_params",
"path_with_sse_slash_and_query_params",
@ -758,7 +760,7 @@ class TestInferTransport:
"url",
[
"http://example.com/api",
"https://localhost:8080/mcp",
"https://localhost:8080/mcp/",
"http://example.com/asset/image.jpg",
"https://localhost:8080/sservice/endpoint",
"https://example.com/assets/file",
@ -779,7 +781,7 @@ class TestInferTransport:
config = {
"mcpServers": {
"test_server": {
"url": "http://localhost:8000/sse",
"url": "http://localhost:8000/sse/",
"headers": {"Authorization": "Bearer 123"},
},
}
@ -787,7 +789,7 @@ class TestInferTransport:
transport = infer_transport(config)
assert isinstance(transport, MCPConfigTransport)
assert isinstance(transport.transport, SSETransport)
assert transport.transport.url == "http://localhost:8000/sse"
assert transport.transport.url == "http://localhost:8000/sse/"
assert transport.transport.headers == {"Authorization": "Bearer 123"}
def test_infer_local_transport_from_config(self):
@ -825,7 +827,7 @@ class TestInferTransport:
"args": ["hello"],
},
"remote": {
"url": "http://localhost:8000/sse",
"url": "http://localhost:8000/sse/",
"headers": {"Authorization": "Bearer 123"},
},
}

6
tests/client/test_openapi.py
View file

@ -57,12 +57,12 @@ class TestClientHeaders:
@pytest.fixture(scope="class")
def shttp_server(self) -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="streamable-http") as url:
yield f"{url}/mcp"
yield f"{url}/mcp/"
@pytest.fixture(scope="class")
def sse_server(self) -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="sse") as url:
yield f"{url}/sse"
yield f"{url}/sse/"
@pytest.fixture(scope="class")
def proxy_server(self, shttp_server: str) -> Generator[str, None, None]:
@ -71,7 +71,7 @@ class TestClientHeaders:
shttp_url=shttp_server,
transport="streamable-http",
) as url:
yield f"{url}/mcp"
yield f"{url}/mcp/"
async def test_client_headers_sse_resource(self, sse_server: str):
async with Client(

6
tests/client/test_sse.py
View file

@ -70,7 +70,7 @@ def run_server(host: str, port: int, **kwargs) -> None:
@pytest.fixture(autouse=True, scope="module")
def sse_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="sse") as url:
yield f"{url}/sse"
yield f"{url}/sse/"
async def test_ping(sse_server: str):
@ -92,7 +92,7 @@ async def test_http_headers(sse_server: str):
def run_nested_server(host: str, port: int) -> None:
app = fastmcp_server().sse_app(path="/mcp/sse", message_path="/mcp/messages")
app = fastmcp_server().sse_app(path="/mcp/sse/", message_path="/mcp/messages")
mount = Starlette(routes=[Mount("/nest-inner", app=app)])
mount2 = Starlette(routes=[Mount("/nest-outer", app=mount)])
server = uvicorn.Server(
@ -114,7 +114,7 @@ async def test_nested_sse_server_resolves_correctly():
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=SSETransport(f"{url}/nest-outer/nest-inner/mcp/sse")
transport=SSETransport(f"{url}/nest-outer/nest-inner/mcp/sse/")
) as client:
result = await client.ping()
assert result is True

8
tests/client/test_streamable_http.py
View file

@ -79,7 +79,7 @@ def run_server(host: str, port: int, stateless_http: bool = False, **kwargs) ->
def run_nested_server(host: str, port: int) -> None:
mcp_app = fastmcp_server().http_app(path="/final/mcp")
mcp_app = fastmcp_server().http_app(path="/final/mcp/")
mount = Starlette(routes=[Mount("/nest-inner", app=mcp_app)])
mount2 = Starlette(
@ -105,9 +105,9 @@ async def streamable_http_server(
with run_server_in_process(
run_server, stateless_http=stateless_http, transport="streamable-http"
) as url:
async with Client(transport=StreamableHttpTransport(f"{url}/mcp")) as client:
async with Client(transport=StreamableHttpTransport(f"{url}/mcp/")) as client:
assert await client.ping()
yield f"{url}/mcp"
yield f"{url}/mcp/"
async def test_ping(streamable_http_server: str):
@ -156,7 +156,7 @@ async def test_nested_streamable_http_server_resolves_correctly():
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/final/mcp")
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/final/mcp/")
) as client:
result = await client.ping()
assert result is True

4
tests/deprecated/test_settings.py
View file

@ -123,7 +123,7 @@ class TestDeprecatedServerInitKwargs:
debug=False,
host="127.0.0.1",
port=9999,
sse_path="/sse",
sse_path="/sse/",
message_path="/msg",
streamable_http_path="/http",
json_response=False,
@ -162,7 +162,7 @@ class TestDeprecatedServerInitKwargs:
assert server._deprecated_settings.debug is False
assert server._deprecated_settings.host == "127.0.0.1"
assert server._deprecated_settings.port == 9999
assert server._deprecated_settings.sse_path == "/sse"
assert server._deprecated_settings.sse_path == "/sse/"
assert server._deprecated_settings.message_path == "/msg"
assert server._deprecated_settings.streamable_http_path == "/http"
assert server._deprecated_settings.json_response is False

2
tests/server/http/test_custom_routes.py
View file

@ -55,7 +55,7 @@ class TestCustomRoutes:
"""Test that custom routes are included when using create_sse_app directly."""
# Create the app by calling the constructor function directly
app = create_sse_app(
server=server_with_custom_route, message_path="/message", sse_path="/sse"
server=server_with_custom_route, message_path="/message", sse_path="/sse/"
)
# Verify that the custom route is included

4
tests/server/http/test_http_dependencies.py
View file

@ -45,13 +45,13 @@ def run_server(host: str, port: int, **kwargs) -> None:
@pytest.fixture(autouse=True, scope="module")
def shttp_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="streamable-http") as url:
yield f"{url}/mcp"
yield f"{url}/mcp/"
@pytest.fixture(autouse=True, scope="module")
def sse_server() -> Generator[str, None, None]:
with run_server_in_process(run_server, transport="sse") as url:
yield f"{url}/sse"
yield f"{url}/sse/"
async def test_http_headers_resource_shttp(shttp_server: str):

2
tests/server/http/test_http_middleware.py
View file

@ -126,7 +126,7 @@ async def test_create_sse_app_with_custom_middleware():
app = create_sse_app(
server=server,
message_path="/message",
sse_path="/sse",
sse_path="/sse/",
middleware=custom_middleware,
routes=additional_routes,
)

4
tests/server/test_app_state.py
View file

@ -16,11 +16,11 @@ def test_http_app_sse_sets_mcp_server_state():
def test_create_streamable_http_app_sets_state():
server = FastMCP(name="StateTest")
app = create_streamable_http_app(server, "/mcp")
app = create_streamable_http_app(server, "/mcp/")
assert app.state.fastmcp_server is server
def test_create_sse_app_sets_state():
server = FastMCP(name="StateTest")
app = create_sse_app(server, message_path="/message", sse_path="/sse")
app = create_sse_app(server, message_path="/message", sse_path="/sse/")
assert app.state.fastmcp_server is server

4
tests/server/test_mount.py
View file

@ -273,7 +273,9 @@ class TestMultipleServerMount:
main_app.mount(working_app, "working")
# Use an unreachable port
unreachable_client = Client(transport=SSETransport("http://127.0.0.1:9999/sse"))
unreachable_client = Client(
transport=SSETransport("http://127.0.0.1:9999/sse/")
)
# Create a proxy server that will fail to connect
unreachable_proxy = FastMCP.as_proxy(unreachable_client)

4
tests/server/test_proxy.py
View file

@ -102,10 +102,10 @@ async def test_as_proxy_with_transport(fastmcp_server):
def test_as_proxy_with_url():
"""FastMCP.as_proxy should accept a URL without connecting."""
proxy = FastMCP.as_proxy("http://example.com/mcp")
proxy = FastMCP.as_proxy("http://example.com/mcp/")
assert isinstance(proxy, FastMCPProxy)
assert isinstance(proxy.client.transport, StreamableHttpTransport)
assert proxy.client.transport.url == "http://example.com/mcp"
assert proxy.client.transport.url == "http://example.com/mcp/"
class TestTools:

8
tests/utilities/test_mcp_config.py
View file

@ -61,21 +61,21 @@ def test_parse_remote_config_with_url_inference():
config = {
"mcpServers": {
"test_server": {
"url": "http://localhost:8000/sse",
"url": "http://localhost:8000/sse/",
}
}
}
mcp_config = MCPConfig.from_dict(config)
transport = mcp_config.mcpServers["test_server"].to_transport()
assert isinstance(transport, SSETransport)
assert transport.url == "http://localhost:8000/sse"
assert transport.url == "http://localhost:8000/sse/"
def test_parse_multiple_servers():
config = {
"mcpServers": {
"test_server": {
"url": "http://localhost:8000/sse",
"url": "http://localhost:8000/sse/",
},
"test_server_2": {
"command": "echo",
@ -172,7 +172,7 @@ async def test_remote_config_sse_with_auth_token():
config = {
"mcpServers": {
"test_server": {
"url": "http://localhost:8000/sse",
"url": "http://localhost:8000/sse/",
"auth": "test_token",
}
}