0
# FastMCP Server Framework
1

2
High-level server framework using decorators for rapid MCP server development. FastMCP provides a modern, intuitive API for building MCP servers with built-in support for tools, resources, prompts, HTTP endpoints, authentication, and multiple transport protocols.
3

4
## Capabilities
5

6
### FastMCP Server Class
7

8
Main server class with decorator-based configuration for tools, resources, prompts, and HTTP endpoints.
9

10
```python { .api }
11
class FastMCP:
12
    def __init__(
13
        self,
14
        name: str | None = None,
15
        instructions: str | None = None,
16
        auth_server_provider: OAuthAuthorizationServerProvider | None = None,
17
        token_verifier: TokenVerifier | None = None,
18
        *,
19
        tools: list[Tool] | None = None,
20
        debug: bool = False,
21
        log_level: str = "INFO",
22
        host: str = "127.0.0.1",
23
        port: int = 8000,
24
        **kwargs
25
    ):
26
        """
27
        Initialize FastMCP server instance.
28

29
        Parameters:
30
        - name: Server name for identification
31
        - instructions: Server instructions/description
32
        - auth_server_provider: OAuth authorization server provider
33
        - token_verifier: Token verification handler
34
        - tools: Pre-defined tools list
35
        - debug: Enable debug mode
36
        - log_level: Logging level (DEBUG, INFO, WARNING, ERROR)
37
        - host: Server host address
38
        - port: Server port number
39
        - **kwargs: Additional configuration options
40
        """
41

42
    def tool(
43
        self,
44
        name: str | None = None,
45
        description: str | None = None,
46
        **kwargs
47
    ) -> Callable:
48
        """
49
        Decorator for registering tool functions.
50

51
        Parameters:
52
        - name: Tool name (defaults to function name)
53
        - description: Tool description (defaults to docstring)
54
        - **kwargs: Additional tool metadata
55

56
        Returns:
57
        Decorator function for tool registration
58
        """
59

60
    def resource(
61
        self,
62
        uri: str | None = None,
63
        name: str | None = None,
64
        description: str | None = None,
65
        mime_type: str | None = None,
66
        **kwargs
67
    ) -> Callable:
68
        """
69
        Decorator for registering resource handlers.
70

71
        Parameters:
72
        - uri: Resource URI pattern
73
        - name: Resource name (defaults to function name)
74
        - description: Resource description (defaults to docstring)
75
        - mime_type: Resource MIME type
76
        - **kwargs: Additional resource metadata
77

78
        Returns:
79
        Decorator function for resource registration
80
        """
81

82
    def prompt(
83
        self,
84
        name: str | None = None,
85
        description: str | None = None,
86
        **kwargs
87
    ) -> Callable:
88
        """
89
        Decorator for registering prompt templates.
90

91
        Parameters:
92
        - name: Prompt name (defaults to function name)
93
        - description: Prompt description (defaults to docstring)
94
        - **kwargs: Additional prompt metadata
95

96
        Returns:
97
        Decorator function for prompt registration
98
        """
99

100
    def get(self, path: str, **kwargs) -> Callable:
101
        """
102
        Decorator for HTTP GET endpoint handlers.
103

104
        Parameters:
105
        - path: URL path pattern
106
        - **kwargs: Additional endpoint options
107

108
        Returns:
109
        Decorator function for endpoint registration
110
        """
111

112
    def post(self, path: str, **kwargs) -> Callable:
113
        """
114
        Decorator for HTTP POST endpoint handlers.
115

116
        Parameters:
117
        - path: URL path pattern
118
        - **kwargs: Additional endpoint options
119

120
        Returns:
121
        Decorator function for endpoint registration
122
        """
123

124
    def put(self, path: str, **kwargs) -> Callable:
125
        """
126
        Decorator for HTTP PUT endpoint handlers.
127

128
        Parameters:
129
        - path: URL path pattern
130
        - **kwargs: Additional endpoint options
131

132
        Returns:
133
        Decorator function for endpoint registration
134
        """
135

136
    def delete(self, path: str, **kwargs) -> Callable:
137
        """
138
        Decorator for HTTP DELETE endpoint handlers.
139

140
        Parameters:
141
        - path: URL path pattern
142
        - **kwargs: Additional endpoint options
143

144
        Returns:
145
        Decorator function for endpoint registration
146
        """
147

148
    async def run_stdio_async(self) -> None:
149
        """
150
        Run the server using stdio transport asynchronously.
151
        """
152

153
    async def run_sse_async(
154
        self,
155
        mount_path: str | None = None,
156
        **kwargs
157
    ) -> None:
158
        """
159
        Run the server using Server-Sent Events transport asynchronously.
160

161
        Parameters:
162
        - mount_path: Optional mount path for SSE endpoint
163
        - **kwargs: Additional server options
164
        """
165

166
    async def run_streamable_http_async(self) -> None:
167
        """
168
        Run the server using streamable HTTP transport asynchronously.
169
        """
170

171
    def run(
172
        self,
173
        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
174
        mount_path: str | None = None,
175
    ) -> None:
176
        """
177
        Run the FastMCP server synchronously.
178

179
        Parameters:
180
        - transport: Transport protocol ("stdio", "sse", or "streamable-http")
181
        - mount_path: Optional mount path for SSE transport
182
        """
183
```
184

185
### Context Access
186

187
Request context object providing access to session information and request metadata within handler functions.
188

189
```python { .api }
190
class Context:
191
    @property
192
    def request_id(self) -> str:
193
        """Current request identifier."""
194

195
    @property
196
    def client_session(self) -> ServerSession:
197
        """Current client session."""
198

199
    @property
200
    def user_id(self) -> str | None:
201
        """Authenticated user ID (if authentication enabled)."""
202

203
    @property
204
    def request_metadata(self) -> dict[str, Any]:
205
        """Request metadata and headers."""
206

207
    async def send_progress(
208
        self,
209
        progress: float,
210
        total: float | None = None,
211
        message: str | None = None
212
    ) -> None:
213
        """
214
        Send progress notification to client.
215

216
        Parameters:
217
        - progress: Current progress value
218
        - total: Total expected value
219
        - message: Progress message
220
        """
221

222
    async def send_log(
223
        self,
224
        level: LoggingLevel,
225
        message: str,
226
        **kwargs
227
    ) -> None:
228
        """
229
        Send log message to client.
230

231
        Parameters:
232
        - level: Log level
233
        - message: Log message
234
        - **kwargs: Additional log data
235
        """
236
```
237

238
### Utility Types
239

240
Additional types and utilities for FastMCP development.
241

242
```python { .api }
243
class Image:
244
    def __init__(
245
        self,
246
        data: bytes,
247
        mime_type: str = "image/png",
248
        **kwargs
249
    ):
250
        """
251
        Image data container for FastMCP.
252

253
        Parameters:
254
        - data: Image data bytes
255
        - mime_type: Image MIME type
256
        - **kwargs: Additional metadata
257
        """
258

259
    @classmethod
260
    def from_file(cls, path: str) -> "Image":
261
        """
262
        Load image from file path.
263

264
        Parameters:
265
        - path: File path to image
266

267
        Returns:
268
        Image instance
269
        """
270

271
    @classmethod
272
    def from_url(cls, url: str) -> "Image":
273
        """
274
        Load image from URL.
275

276
        Parameters:
277
        - url: Image URL
278

279
        Returns:
280
        Image instance
281
        """
282

283
    def to_base64(self) -> str:
284
        """
285
        Convert image to base64 string.
286

287
        Returns:
288
        Base64 encoded image data
289
        """
290
```
291

292
## Usage Examples
293

294
### Basic FastMCP Server
295

296
```python
297
from mcp.server import FastMCP
298
import asyncio
299

300
# Create server instance
301
app = FastMCP("example-server", instructions="An example MCP server")
302

303
@app.tool()
304
async def calculate(operation: str, a: float, b: float) -> float:
305
    """Perform basic mathematical operations."""
306
    if operation == "add":
307
        return a + b
308
    elif operation == "subtract":
309
        return a - b
310
    elif operation == "multiply":
311
        return a * b
312
    elif operation == "divide":
313
        if b == 0:
314
            raise ValueError("Cannot divide by zero")
315
        return a / b
316
    else:
317
        raise ValueError(f"Unknown operation: {operation}")
318

319
@app.resource("config://settings")
320
async def get_settings() -> str:
321
    """Get application configuration."""
322
    return "debug=true\nlog_level=INFO\nmax_connections=100"
323

324
@app.prompt()
325
async def code_review_prompt(language: str, code: str) -> str:
326
    """Generate code review prompt."""
327
    return f"""Please review this {language} code:
328

329
```{language}
330
{code}
331
```
332

333
Focus on:
334
- Code quality and best practices
335
- Potential bugs or issues
336
- Performance considerations
337
- Security concerns
338
"""
339

340
# Run the server
341
if __name__ == "__main__":
342
    app.run("stdio")
343
```
344
345
### Server with HTTP Endpoints
346

347
```python
348
from mcp.server import FastMCP, Context
349
from fastapi import HTTPException
350

351
app = FastMCP("api-server")
352

353
@app.get("/health")
354
async def health_check():
355
    """Health check endpoint."""
356
    return {"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}
357

358
@app.post("/data")
359
async def process_data(data: dict):
360
    """Process submitted data."""
361
    if not data:
362
        raise HTTPException(status_code=400, detail="No data provided")
363
    
364
    # Process the data
365
    result = {"processed": True, "items": len(data)}
366
    return result
367

368
@app.tool()
369
async def get_api_status(ctx: Context) -> dict:
370
    """Get current API server status."""
371
    return {
372
        "request_id": ctx.request_id,
373
        "server": "api-server",
374
        "status": "running"
375
    }
376

377
# Run with HTTP transport
378
if __name__ == "__main__":
379
    app.run("sse")
380
```
381

382
### Server with Authentication
383

384
```python
385
from mcp.server import FastMCP, Context
386
from mcp.server.auth import AuthSettings, ProviderTokenVerifier
387

388
# Configure authentication
389
auth_settings = AuthSettings(
390
    client_id="your-client-id",
391
    client_secret="your-client-secret",  
392
    authorization_endpoint="https://auth.example.com/oauth/authorize",
393
    token_endpoint="https://auth.example.com/oauth/token"
394
)
395

396
token_verifier = ProviderTokenVerifier(auth_settings)
397

398
app = FastMCP(
399
    "secure-server",
400
    auth_server_provider=auth_settings,
401
    token_verifier=token_verifier
402
)
403

404
@app.tool()
405
async def get_user_data(ctx: Context) -> dict:
406
    """Get data for authenticated user."""
407
    user_id = ctx.user_id
408
    if not user_id:
409
        raise ValueError("Authentication required")
410
    
411
    return {
412
        "user_id": user_id,
413
        "data": f"User data for {user_id}"
414
    }
415

416
@app.resource("user://profile")
417
async def user_profile(ctx: Context) -> str:
418
    """Get user profile information."""
419
    user_id = ctx.user_id
420
    if not user_id:
421
        raise ValueError("Authentication required")
422
        
423
    return f"Profile data for user: {user_id}"
424

425
if __name__ == "__main__":
426
    app.run("sse")
427
```
428

429
### Progress Reporting
430

431
```python
432
from mcp.server import FastMCP, Context
433
import asyncio
434

435
app = FastMCP("progress-server")
436

437
@app.tool()
438
async def long_running_task(ctx: Context, iterations: int = 100) -> str:
439
    """Demonstrate progress reporting during long operations."""
440
    for i in range(iterations):
441
        # Simulate work
442
        await asyncio.sleep(0.1)
443
        
444
        # Report progress
445
        await ctx.send_progress(
446
            progress=i + 1,
447
            total=iterations,
448
            message=f"Processing item {i + 1}/{iterations}"
449
        )
450
    
451
    return f"Completed {iterations} iterations"
452

453
@app.tool()  
454
async def file_processor(ctx: Context, files: list[str]) -> dict:
455
    """Process multiple files with progress tracking."""
456
    results = {}
457
    
458
    for idx, filename in enumerate(files):
459
        await ctx.send_progress(
460
            progress=idx,
461
            total=len(files), 
462
            message=f"Processing {filename}"
463
        )
464
        
465
        # Simulate file processing
466
        await asyncio.sleep(0.5)
467
        results[filename] = f"Processed {filename}"
468
    
469
    return {"processed_files": results, "total": len(files)}
470

471
if __name__ == "__main__":
472
    app.run("stdio")
473
```
474

475
### Resource Templates and Dynamic URIs
476

477
```python
478
from mcp.server import FastMCP
479
import os
480

481
app = FastMCP("file-server")
482

483
@app.resource("file://{path}")
484
async def read_file(path: str) -> str:
485
    """Read file content by path."""
486
    if not os.path.exists(path):
487
        raise FileNotFoundError(f"File not found: {path}")
488
    
489
    with open(path, 'r') as f:
490
        return f.read()
491

492
@app.resource("config://{section}/{key}")
493
async def get_config_value(section: str, key: str) -> str:
494
    """Get configuration value by section and key."""
495
    config = {
496
        "database": {"host": "localhost", "port": "5432"},
497
        "api": {"key": "secret", "timeout": "30"}
498
    }
499
    
500
    if section not in config:
501
        raise ValueError(f"Unknown section: {section}")
502
    
503
    if key not in config[section]:
504
        raise ValueError(f"Unknown key: {key}")
505
    
506
    return config[section][key]
507

508
@app.tool()
509
async def list_files(directory: str = ".") -> list[str]:
510
    """List files in a directory."""
511
    if not os.path.isdir(directory):
512
        raise ValueError(f"Not a directory: {directory}")
513
    
514
    return [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]
515

516
if __name__ == "__main__":
517
    app.run("stdio")
518
```