Description
Mounting an FastMCP http_app leads to errors when testing the FastAPI endpoints using fastapi.testclient.TestClient.
Consider the following minimal app in example_api/app.py:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Define FastAPI lifespan."""
print('Starting FastAPI app')
try:
yield
finally:
print('Stopping FastAPI app')
mcp = FastMCP('tools')
mcp_app = mcp.http_app(path='/mcp', transport='http')
@mcp.tool
def add_two_numbers(first: float, second: float) -> float:
"""Add two numbers."""
return first + second
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
"""Run both lifespans."""
async with lifespan(app):
async with mcp_app.lifespan(app):
yield
app = FastAPI(lifespan=combined_lifespan)
@app.get('/health')
def health():
"""Return health status."""
return {'status': 'ok'}
app.mount('/analytics', mcp_app)and a test like this:
from fastapi.testclient import TestClient
from example_app.app import app
def test_health():
with TestClient(app) as client:
response = client.get('/health')
assert response.status_code == 200
def test_health2():
with TestClient(app) as client:
response = client.get('/health')
assert response.status_code == 200Running the tests with pytest test_app.py leads to the following error:
RuntimeError: StreamableHTTPSessionManager .run() can only be called once per instance. Create a new instance if you need to run again.
The bug has already been found in this issue: #946 and in a linked blog a solution was proposed. Changing the app to following fixes the error:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP
class CustomFastMCP():
def __init__(self, mcp):
self.mcp = mcp
def create_app(self):
self.app = self.mcp.http_app(path='/mcp', transport='http')
self.generator = self.app.lifespan(self.app)
self.generator_started = False
return self.app
async def start_session(self):
if self.generator_started:
return
await self.generator.__aenter__()
self.generator_started = True
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Define FastAPI lifespan."""
print('Starting FastAPI app')
try:
yield
finally:
print('Stopping FastAPI app')
mcp = FastMCP('tools')
custom = CustomFastMCP(mcp)
@mcp.tool
def add_two_numbers(first: float, second: float) -> float:
"""Add two numbers."""
return first + second
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
"""Run both lifespans."""
async with lifespan(app):
await custom.start_session()
yield
app = FastAPI(lifespan=combined_lifespan)
@app.get('/health')
def health():
"""Return health status."""
return {'status': 'ok'}
app.mount('/analytics', custom.create_app())Example Code
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastmcp import FastMCP
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Define FastAPI lifespan."""
print('Starting FastAPI app')
try:
yield
finally:
print('Stopping FastAPI app')
mcp = FastMCP('tools')
mcp_app = mcp.http_app(path='/mcp', transport='http')
@mcp.tool
def add_two_numbers(first: float, second: float) -> float:
"""Add two numbers."""
return first + second
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
"""Run both lifespans."""
async with lifespan(app):
async with mcp_app.lifespan(app):
yield
app = FastAPI(lifespan=combined_lifespan)
@app.get('/health')
def health():
"""Return health status."""
return {'status': 'ok'}
app.mount('/analytics', mcp_app)
def test_health():
with TestClient(app) as client:
response = client.get('/health')
assert response.status_code == 200
def test_health2():
with TestClient(app) as client:
response = client.get('/health')
assert response.status_code == 200Version Information
FastMCP version: 2.13.0.2
MCP version: 1.20.0
Python version: 3.10.18
Platform: Windows-10-10.0.19044-SP0
Description
Mounting an FastMCP
http_appleads to errors when testing the FastAPI endpoints usingfastapi.testclient.TestClient.Consider the following minimal app in
example_api/app.py:and a test like this:
Running the tests with
pytest test_app.pyleads to the following error:RuntimeError: StreamableHTTPSessionManager .run() can only be called once per instance. Create a new instance if you need to run again.The bug has already been found in this issue: #946 and in a linked blog a solution was proposed. Changing the app to following fixes the error:
Example Code
Version Information