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

allow passing one or many --server-arg flags

Browse source
This commit is contained in:
zzstoatzz 2025-06-03 11:02:52 -05:00
commit dc88356869
6 changed files with 167 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -61,6 +61,19 @@ fastmcp dev server.py
See the [CLI documentation](/patterns/cli) for detailed information about all available commands and options.
### Passing Arguments to Servers
<VersionBadge version="2.6.2" />
When servers accept command line arguments (using argparse, click, or other libraries), you can pass them using the `--server-arg` option:
```bash
fastmcp run config_server.py --server-arg="--config" --server-arg="config.json"
fastmcp run database_server.py --server-arg="--database-path" --server-arg="/tmp/db.sqlite"
```
This is useful for servers that need configuration files, database paths, API keys, or other runtime options.
## Transport Options
Below is a comparison of available transport options to help you choose the right one for your needs:
@ -157,117 +170,4 @@ 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/`).
<CodeGroup>
```python {6} server.py
from fastmcp import FastMCP
mcp = FastMCP()
if __name__ == "__main__":
mcp.run(transport="sse")
```
```python {3,7} client.py
import asyncio
from fastmcp import Client
from fastmcp.client.transports import SSETransport
async def example():
async with Client(
transport=SSETransport("http://127.0.0.1:8000/sse")
) as client:
await client.ping()
if __name__ == "__main__":
asyncio.run(example())
```
</CodeGroup>
<Tip>
Notice that the client in the above example uses an explicit `SSETransport` to connect to the server. FastMCP will attempt to infer the appropriate transport from the provided configuration, but HTTP URLs are assumed to be Streamable HTTP (as of FastMCP 2.3.0).
</Tip>
To customize the host, port, or log level, provide appropriate keyword arguments to the `run()` method. You can also adjust the SSE path (which clients should connect to) and the message POST endpoint (which clients use to send subsequent messages).
<CodeGroup>
```python {8-12} server.py
from fastmcp import FastMCP
mcp = FastMCP()
if __name__ == "__main__":
mcp.run(
transport="sse",
host="127.0.0.1",
port=4200,
log_level="debug",
path="/my-custom-sse-path",
)
```
```python {7} client.py
import asyncio
from fastmcp import Client
from fastmcp.client.transports import SSETransport
async def example():
async with Client(
transport=SSETransport("http://127.0.0.1:4200/my-custom-sse-path")
) as client:
await client.ping()
if __name__ == "__main__":
asyncio.run(example())
```
</CodeGroup>
## Async Usage
FastMCP provides both synchronous and asynchronous APIs for running your server. The `run()` method seen in previous examples is a synchronous method that internally uses `anyio.run()` to run the asynchronous server. For applications that are already running in an async context, FastMCP provides the `run_async()` method.
```python {10-12}
from fastmcp import FastMCP
import asyncio
mcp = FastMCP(name="MyServer")
@mcp.tool()
def hello(name: str) -> str:
return f"Hello, {name}!"
async def main():
# Use run_async() in async contexts
await mcp.run_async(transport="streamable-http")
if __name__ == "__main__":
asyncio.run(main())
```
<Warning>
The `run()` method cannot be called from inside an async function because it already creates its own async event loop internally. If you attempt to call `run()` from inside an async function, you'll get an error about the event loop already running.
Always use `run_async()` inside async functions and `run()` in synchronous contexts.
</Warning>
Both `run()` and `run_async()` accept the same transport arguments, so all the examples above apply to both methods.
## Custom Routes
You can also add custom web routes to your FastMCP server, which will be exposed alongside the MCP endpoint. To do so, use the `@custom_route` decorator. Note that this is less flexible than using a full ASGI framework, but can be useful for adding simple endpoints like health checks to your standalone server.
```python
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import PlainTextResponse
mcp = FastMCP("MyServer")
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> PlainTextResponse:
return PlainTextResponse("OK")
if __name__ == "__main__":
mcp.run()
```
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/`

47
examples/config_server.py Normal file
View file

@ -0,0 +1,47 @@
"""
Simple example showing FastMCP server with command line argument support.
Usage:
fastmcp run examples/config_server.py --server-arg="--name" --server-arg="MyServer"
fastmcp run examples/config_server.py --server-arg="--debug"
"""
import argparse
from fastmcp import FastMCP
parser = argparse.ArgumentParser(description="Simple configurable MCP server")
parser.add_argument(
"--name", type=str, default="ConfigurableServer", help="Server name"
)
parser.add_argument("--debug", action="store_true", help="Enable debug mode")
args = parser.parse_args()
server_name = args.name
if args.debug:
server_name += " (Debug)"
mcp = FastMCP(server_name)
@mcp.tool()
def get_status() -> dict[str, str | bool]:
"""Get the current server configuration and status."""
return {
"server_name": server_name,
"debug_mode": args.debug,
"original_name": args.name,
}
@mcp.tool()
def echo_message(message: str) -> str:
"""Echo a message, with debug info if debug mode is enabled."""
if args.debug:
return f"[DEBUG] Echoing: {message}"
return message
if __name__ == "__main__":
mcp.run()

12
src/fastmcp/cli/cli.py
View file

@ -256,6 +256,13 @@ def run(
help="Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
),
] = None,
server_args: Annotated[
list[str],
typer.Option(
"--server-arg",
help="Additional arguments to pass to the server",
),
] = [],
) -> None:
"""Run a MCP server or connect to a remote one.
@ -266,6 +273,9 @@ def run(
Note: This command runs the server directly. You are responsible for ensuring
all dependencies are available.
Server arguments can be passed using --server-arg:
fastmcp run server.py --server-arg="--config" --server-arg="config.json"
"""
logger.debug(
"Running server or client",
@ -275,6 +285,7 @@ def run(
"host": host,
"port": port,
"log_level": log_level,
"server_args": server_args,
},
)
@ -285,6 +296,7 @@ def run(
host=host,
port=port,
log_level=log_level,
server_args=server_args,
)
except Exception as e:
logger.error(

33
src/fastmcp/cli/run.py
View file

@ -71,6 +71,9 @@ def import_server(file: Path, server_object: str | None = None) -> Any:
logger.error("Could not load module", extra={"file": str(file)})
sys.exit(1)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
@ -89,6 +92,8 @@ def import_server(file: Path, server_object: str | None = None) -> Any:
)
sys.exit(1)
assert server_object is not None
# Handle module:object syntax
if ":" in server_object:
module_name, object_name = server_object.split(":", 1)
@ -135,12 +140,37 @@ def create_client_server(url: str) -> Any:
sys.exit(1)
def import_server_with_args(
file: Path, server_object: str | None = None, server_args: list[str] | None = None
) -> Any:
"""Import a server with optional command line arguments.
Args:
file: Path to the server file
server_object: Optional server object name
server_args: Optional command line arguments to inject
Returns:
The imported server object
"""
if server_args:
original_argv = sys.argv[:]
try:
sys.argv = [str(file)] + server_args
return import_server(file, server_object)
finally:
sys.argv = original_argv
else:
return import_server(file, server_object)
def run_command(
server_spec: str,
transport: str | None = None,
host: str | None = None,
port: int | None = None,
log_level: str | None = None,
server_args: list[str] | None = None,
) -> None:
"""Run a MCP server or connect to a remote one.
@ -150,6 +180,7 @@ def run_command(
host: Host to bind to when using http transport
port: Port to bind to when using http transport
log_level: Log level
server_args: Additional arguments to pass to the server
"""
if is_url(server_spec):
# Handle URL case
@ -158,7 +189,7 @@ def run_command(
else:
# Handle file case
file, server_object = parse_file_path(server_spec)
server = import_server(file, server_object)
server = import_server_with_args(file, server_object, server_args)
logger.debug(f'Found server "{server.name}" in {file}')
# Run the server

26
tests/cli/test_cli.py
View file

@ -409,3 +409,29 @@ class TestRunCommand:
mock_server.run.assert_called_once_with(
transport="sse", host="0.0.0.0", port=8080, log_level="DEBUG"
)
def test_run_command_with_server_args(self, temp_python_file):
"""Test run command with server arguments."""
with (
patch("fastmcp.cli.run.run_command") as mock_run_command,
):
result = runner.invoke(
cli.app,
[
"run",
str(temp_python_file),
"--server-arg",
"--config",
"--server-arg",
"config.json",
],
)
assert result.exit_code == 0
mock_run_command.assert_called_once_with(
server_spec=str(temp_python_file),
transport=None,
host=None,
port=None,
log_level=None,
server_args=["--config", "config.json"],
)

36
tests/cli/test_run.py
View file

@ -260,3 +260,39 @@ class TestRunCommand:
mock_server.run.assert_called_once_with(
transport="sse", host="0.0.0.0", port=8080, log_level="DEBUG"
)
class TestImportServerWithArgs:
"""Tests for the import_server_with_args function."""
def test_import_server_with_args_no_args(self, temp_python_file):
"""Test importing server without arguments."""
with patch("fastmcp.cli.run.import_server") as mock_import:
mock_server = MagicMock()
mock_import.return_value = mock_server
result = fastmcp.cli.run.import_server_with_args(
temp_python_file, None, None
)
assert result == mock_server
mock_import.assert_called_once_with(temp_python_file, None)
def test_import_server_with_args_with_args(self, temp_python_file):
"""Test importing server with arguments."""
import sys
with patch("fastmcp.cli.run.import_server") as mock_import:
mock_server = MagicMock()
mock_import.return_value = mock_server
original_argv = sys.argv[:]
result = fastmcp.cli.run.import_server_with_args(
temp_python_file, "custom_server", ["--config", "test.json", "--debug"]
)
assert result == mock_server
mock_import.assert_called_once_with(temp_python_file, "custom_server")
# Verify sys.argv was restored
assert sys.argv == original_argv