1
# Utilities and Types
2

3
Helper classes and utility functions for enhanced functionality and type support. FastMCP provides a comprehensive set of utilities for common tasks, data handling, and system integration.
4

5
## Capabilities
6

7
### Media and File Types
8

9
Helper classes for returning rich media content from tools and resources.
10

11
```python { .api }
12
class Image:
13
    def __init__(
14
        self,
15
        data: bytes | str,
16
        mime_type: str = "image/png"
17
    ):
18
        """
19
        Image helper for returning image data from tools.
20
        
21
        Parameters:
22
        - data: Image data as bytes or base64 string
23
        - mime_type: MIME type of the image
24
        """
25

26
class Audio:
27
    def __init__(
28
        self,
29
        data: bytes | str,
30
        mime_type: str = "audio/wav"
31
    ):
32
        """
33
        Audio helper for returning audio data from tools.
34
        
35
        Parameters:
36
        - data: Audio data as bytes or base64 string
37
        - mime_type: MIME type of the audio
38
        """
39

40
class File:
41
    def __init__(
42
        self,
43
        data: bytes | str,
44
        name: str,
45
        mime_type: str | None = None
46
    ):
47
        """
48
        File helper for returning file data from tools.
49
        
50
        Parameters:
51
        - data: File content as bytes or string
52
        - name: Filename
53
        - mime_type: MIME type of the file
54
        """
55
```
56

57
### Base Model Classes
58

59
Base Pydantic models providing common functionality for FastMCP components.
60

61
```python { .api }
62
class FastMCPBaseModel:
63
    """Base Pydantic model for FastMCP with common configurations."""
64
    
65
    class Config:
66
        arbitrary_types_allowed = True
67
        extra = "forbid"
68
        validate_assignment = True
69

70
def get_cached_typeadapter(type_hint: type) -> TypeAdapter:
71
    """
72
    Get cached TypeAdapter for type validation.
73
    
74
    Parameters:
75
    - type_hint: Type to create adapter for
76
    
77
    Returns:
78
    Cached TypeAdapter instance
79
    """
80
```
81

82
### Component System
83

84
Base classes for creating reusable FastMCP components.
85

86
```python { .api }
87
class FastMCPComponent:
88
    """Base class for FastMCP components with lifecycle management."""
89
    
90
    def __init__(self, name: str | None = None):
91
        """
92
        Initialize component.
93
        
94
        Parameters:
95
        - name: Component name for identification
96
        """
97
    
98
    async def initialize(self) -> None:
99
        """Initialize component resources."""
100
    
101
    async def cleanup(self) -> None:
102
        """Clean up component resources."""
103
    
104
    def get_name(self) -> str:
105
        """Get component name."""
106
```
107

108
### Settings Management
109

110
Comprehensive settings system for configuring FastMCP behavior.
111

112
```python { .api }
113
class Settings:
114
    """Main settings class for FastMCP configuration."""
115
    
116
    # Logging settings
117
    log_enabled: bool = True
118
    log_level: str = "INFO"
119
    enable_rich_tracebacks: bool = True
120
    
121
    # Development settings
122
    test_mode: bool = False
123
    debug_mode: bool = False
124
    
125
    # Feature flags
126
    deprecation_warnings: bool = True
127
    experimental_features: bool = False
128
    
129
    # Performance settings
130
    max_connections: int = 100
131
    request_timeout: float = 30.0
132
    
133
    # Security settings
134
    allow_dangerous_tools: bool = False
135
    require_auth: bool = False
136

137
class ExperimentalSettings:
138
    """Settings for experimental features."""
139
    
140
    enable_async_tools: bool = False
141
    enable_streaming: bool = False
142
    enable_caching: bool = False
143
```
144

145
### OpenAPI Integration
146

147
Utilities for working with OpenAPI specifications and HTTP APIs.
148

149
```python { .api }
150
class OpenAPIParser:
151
    """Parser for OpenAPI specifications."""
152
    
153
    def __init__(self, spec: dict | str):
154
        """
155
        Initialize OpenAPI parser.
156
        
157
        Parameters:
158
        - spec: OpenAPI specification as dict or JSON string
159
        """
160
    
161
    def parse_paths(self) -> list[HTTPRoute]:
162
        """Parse API paths into HTTPRoute objects."""
163
    
164
    def get_schemas(self) -> dict[str, dict]:
165
        """Get schema definitions from specification."""
166

167
class HTTPRoute:
168
    """HTTP route representation."""
169
    
170
    def __init__(
171
        self,
172
        path: str,
173
        method: str,
174
        operation_id: str | None = None,
175
        summary: str | None = None,
176
        description: str | None = None,
177
        parameters: list[dict] | None = None,
178
        request_body: dict | None = None,
179
        responses: dict | None = None
180
    ):
181
        """
182
        HTTP route definition.
183
        
184
        Parameters:
185
        - path: URL path with parameters
186
        - method: HTTP method (GET, POST, etc.)
187
        - operation_id: Unique operation identifier
188
        - summary: Brief operation summary
189
        - description: Detailed operation description
190
        - parameters: Request parameters
191
        - request_body: Request body schema
192
        - responses: Response schemas
193
        """
194
```
195

196
### MCP Configuration
197

198
Configuration management for MCP server setups and client connections.
199

200
```python { .api }
201
class MCPConfig:
202
    """MCP configuration format for server definitions."""
203
    
204
    def __init__(self, config: dict):
205
        """
206
        Initialize MCP configuration.
207
        
208
        Parameters:
209
        - config: Configuration dictionary
210
        """
211
    
212
    def get_servers(self) -> dict[str, dict]:
213
        """Get server configurations."""
214
    
215
    def validate(self) -> bool:
216
        """Validate configuration format."""
217
```
218

219
### Testing Utilities
220

221
Helper functions and classes for testing FastMCP servers and clients.
222

223
```python { .api }
224
def create_test_server() -> FastMCP:
225
    """Create a FastMCP server configured for testing."""
226

227
def create_test_client(server: FastMCP) -> Client:
228
    """Create a client connected to test server via in-memory transport."""
229

230
async def assert_tool_exists(client: Client, tool_name: str) -> None:
231
    """Assert that a tool exists on the server."""
232

233
async def assert_resource_exists(client: Client, resource_uri: str) -> None:
234
    """Assert that a resource exists on the server."""
235
```
236

237
### JSON Schema Utilities
238

239
Utilities for working with JSON schemas and type validation.
240

241
```python { .api }
242
def generate_schema_from_function(func: Callable) -> dict:
243
    """
244
    Generate JSON schema from function signature.
245
    
246
    Parameters:
247
    - func: Function to analyze
248
    
249
    Returns:
250
    JSON schema dictionary
251
    """
252

253
def validate_against_schema(data: Any, schema: dict) -> bool:
254
    """
255
    Validate data against JSON schema.
256
    
257
    Parameters:
258
    - data: Data to validate
259
    - schema: JSON schema
260
    
261
    Returns:
262
    True if valid, False otherwise
263
    """
264
```
265

266
### HTTP Utilities
267

268
Utilities for HTTP operations and request handling.
269

270
```python { .api }
271
async def make_http_request(
272
    method: str,
273
    url: str,
274
    headers: dict | None = None,
275
    data: Any | None = None,
276
    timeout: float = 30.0
277
) -> dict:
278
    """
279
    Make HTTP request with error handling.
280
    
281
    Parameters:
282
    - method: HTTP method
283
    - url: Target URL
284
    - headers: Request headers
285
    - data: Request data
286
    - timeout: Request timeout
287
    
288
    Returns:
289
    Response data
290
    """
291

292
def parse_http_headers(headers: dict) -> dict[str, str]:
293
    """Parse and normalize HTTP headers."""
294
```
295

296
## Usage Examples
297

298
### Media Types Usage
299

300
```python
301
from fastmcp import FastMCP
302
from fastmcp.utilities.types import Image, Audio, File
303
import base64
304

305
mcp = FastMCP("Media Server")
306

307
@mcp.tool
308
def create_chart() -> Image:
309
    """Create a chart and return as image."""
310
    import matplotlib.pyplot as plt
311
    import io
312
    
313
    # Create chart
314
    plt.figure(figsize=(10, 6))
315
    plt.plot([1, 2, 3, 4], [1, 4, 2, 3])
316
    plt.title("Sample Chart")
317
    
318
    # Save to bytes
319
    buffer = io.BytesIO()
320
    plt.savefig(buffer, format='png')
321
    buffer.seek(0)
322
    
323
    return Image(buffer.read(), mime_type="image/png")
324

325
@mcp.tool
326
def generate_audio() -> Audio:
327
    """Generate audio data."""
328
    import numpy as np
329
    import wave
330
    import io
331
    
332
    # Generate sine wave
333
    sample_rate = 44100
334
    duration = 1.0
335
    frequency = 440.0
336
    
337
    t = np.linspace(0, duration, int(sample_rate * duration))
338
    waveform = np.sin(2 * np.pi * frequency * t)
339
    
340
    # Convert to audio bytes
341
    buffer = io.BytesIO()
342
    with wave.open(buffer, 'wb') as wav_file:
343
        wav_file.setnchannels(1)
344
        wav_file.setsampwidth(2)
345
        wav_file.setframerate(sample_rate)
346
        wav_file.writeframes((waveform * 32767).astype(np.int16).tobytes())
347
    
348
    buffer.seek(0)
349
    return Audio(buffer.read(), mime_type="audio/wav")
350

351
@mcp.tool
352
def create_document(content: str, filename: str = "document.txt") -> File:
353
    """Create document file."""
354
    return File(
355
        data=content.encode('utf-8'),
356
        name=filename,
357
        mime_type="text/plain"
358
    )
359

360
@mcp.tool
361
def create_json_file(data: dict, filename: str = "data.json") -> File:
362
    """Create JSON file."""
363
    import json
364
    
365
    json_content = json.dumps(data, indent=2)
366
    
367
    return File(
368
        data=json_content,
369
        name=filename,
370
        mime_type="application/json"
371
    )
372
```
373

374
### Settings Configuration
375

376
```python
377
from fastmcp import FastMCP, Settings
378
from fastmcp.utilities.logging import configure_logging
379

380
# Create custom settings
381
settings = Settings()
382
settings.log_level = "DEBUG"
383
settings.debug_mode = True
384
settings.max_connections = 50
385
settings.request_timeout = 60.0
386
settings.allow_dangerous_tools = False
387

388
# Configure logging
389
configure_logging(
390
    level=settings.log_level,
391
    enable_rich_tracebacks=settings.enable_rich_tracebacks
392
)
393

394
# Create server with custom settings
395
mcp = FastMCP(
396
    name="Configured Server",
397
    settings=settings
398
)
399

400
@mcp.tool
401
def debug_info() -> dict:
402
    """Get debug information."""
403
    return {
404
        "debug_mode": settings.debug_mode,
405
        "log_level": settings.log_level,
406
        "max_connections": settings.max_connections,
407
        "request_timeout": settings.request_timeout
408
    }
409

410
mcp.run()
411
```
412

413
### Component System Usage
414

415
```python
416
from fastmcp import FastMCP
417
from fastmcp.utilities.components import FastMCPComponent
418
import asyncio
419

420
class DatabaseComponent(FastMCPComponent):
421
    """Database connection component."""
422
    
423
    def __init__(self):
424
        super().__init__("database")
425
        self.connection = None
426
    
427
    async def initialize(self):
428
        """Initialize database connection."""
429
        # Mock database connection
430
        await asyncio.sleep(0.1)
431
        self.connection = {"status": "connected", "pool_size": 10}
432
        print(f"Component {self.get_name()} initialized")
433
    
434
    async def cleanup(self):
435
        """Clean up database connection."""
436
        if self.connection:
437
            self.connection = None
438
            print(f"Component {self.get_name()} cleaned up")
439
    
440
    def query(self, sql: str) -> list[dict]:
441
        """Execute database query."""
442
        if not self.connection:
443
            raise RuntimeError("Database not connected")
444
        
445
        # Mock query result
446
        return [{"id": 1, "name": "example", "sql": sql}]
447

448
class CacheComponent(FastMCPComponent):
449
    """Cache component."""
450
    
451
    def __init__(self):
452
        super().__init__("cache")
453
        self.cache = {}
454
    
455
    async def initialize(self):
456
        """Initialize cache."""
457
        self.cache = {}
458
        print(f"Component {self.get_name()} initialized")
459
    
460
    async def cleanup(self):
461
        """Clean up cache."""
462
        self.cache.clear()
463
        print(f"Component {self.get_name()} cleaned up")
464
    
465
    def get(self, key: str) -> Any:
466
        """Get value from cache."""
467
        return self.cache.get(key)
468
    
469
    def set(self, key: str, value: Any) -> None:
470
        """Set value in cache."""
471
        self.cache[key] = value
472

473
# Set up components
474
db_component = DatabaseComponent()
475
cache_component = CacheComponent()
476

477
mcp = FastMCP("Component Server")
478

479
@mcp.tool
480
async def query_with_cache(sql: str) -> list[dict]:
481
    """Query database with caching."""
482
    # Check cache first
483
    cached_result = cache_component.get(sql)
484
    if cached_result:
485
        return {"cached": True, "data": cached_result}
486
    
487
    # Query database
488
    result = db_component.query(sql)
489
    
490
    # Cache result
491
    cache_component.set(sql, result)
492
    
493
    return {"cached": False, "data": result}
494

495
# Initialize components before running server
496
async def main():
497
    await db_component.initialize()
498
    await cache_component.initialize()
499
    
500
    try:
501
        mcp.run()
502
    finally:
503
        await db_component.cleanup()
504
        await cache_component.cleanup()
505

506
if __name__ == "__main__":
507
    asyncio.run(main())
508
```
509

510
### OpenAPI Integration
511

512
```python
513
from fastmcp import FastMCP
514
from fastmcp.utilities.openapi import OpenAPIParser, HTTPRoute
515

516
# Load OpenAPI specification
517
openapi_spec = {
518
    "openapi": "3.0.0",
519
    "info": {"title": "Example API", "version": "1.0.0"},
520
    "paths": {
521
        "/users": {
522
            "get": {
523
                "operationId": "getUsers",
524
                "summary": "Get all users",
525
                "responses": {
526
                    "200": {
527
                        "description": "List of users",
528
                        "content": {
529
                            "application/json": {
530
                                "schema": {
531
                                    "type": "array",
532
                                    "items": {"$ref": "#/components/schemas/User"}
533
                                }
534
                            }
535
                        }
536
                    }
537
                }
538
            },
539
            "post": {
540
                "operationId": "createUser",
541
                "summary": "Create new user",
542
                "requestBody": {
543
                    "content": {
544
                        "application/json": {
545
                            "schema": {"$ref": "#/components/schemas/User"}
546
                        }
547
                    }
548
                },
549
                "responses": {
550
                    "201": {"description": "User created"}
551
                }
552
            }
553
        }
554
    },
555
    "components": {
556
        "schemas": {
557
            "User": {
558
                "type": "object",
559
                "properties": {
560
                    "id": {"type": "integer"},
561
                    "name": {"type": "string"},
562
                    "email": {"type": "string"}
563
                }
564
            }
565
        }
566
    }
567
}
568

569
# Parse OpenAPI spec
570
parser = OpenAPIParser(openapi_spec)
571
routes = parser.parse_paths()
572
schemas = parser.get_schemas()
573

574
mcp = FastMCP("OpenAPI Server")
575

576
# Create tools from OpenAPI routes
577
for route in routes:
578
    if route.method == "GET" and route.path == "/users":
579
        @mcp.tool
580
        def get_users() -> list[dict]:
581
            """Get all users."""
582
            return [
583
                {"id": 1, "name": "Alice", "email": "alice@example.com"},
584
                {"id": 2, "name": "Bob", "email": "bob@example.com"}
585
            ]
586
    
587
    elif route.method == "POST" and route.path == "/users":
588
        @mcp.tool
589
        def create_user(name: str, email: str) -> dict:
590
            """Create new user."""
591
            return {
592
                "id": 123,
593
                "name": name,
594
                "email": email,
595
                "created": "2024-01-01T00:00:00Z"
596
            }
597

598
mcp.run()
599
```
600

601
### Testing Utilities Usage
602

603
```python
604
from fastmcp import FastMCP, Client
605
from fastmcp.utilities.tests import create_test_client, assert_tool_exists
606
import asyncio
607

608
async def test_server():
609
    """Test FastMCP server functionality."""
610
    # Create test server
611
    mcp = FastMCP("Test Server")
612
    
613
    @mcp.tool
614
    def add(a: int, b: int) -> int:
615
        """Add two numbers."""
616
        return a + b
617
    
618
    @mcp.resource("config://test")
619
    def get_test_config():
620
        """Get test configuration."""
621
        return {"test": True, "env": "testing"}
622
    
623
    # Create test client
624
    client = create_test_client(mcp)
625
    
626
    async with client:
627
        # Test tool existence
628
        await assert_tool_exists(client, "add")
629
        
630
        # Test tool functionality
631
        result = await client.call_tool("add", {"a": 5, "b": 3})
632
        assert result.text == "8", f"Expected 8, got {result.text}"
633
        
634
        # Test resource
635
        await assert_resource_exists(client, "config://test")
636
        resource = await client.read_resource("config://test")
637
        assert "test" in resource.content
638
        
639
        print("All tests passed!")
640

641
if __name__ == "__main__":
642
    asyncio.run(test_server())
643
```
644

645
### Custom Utility Functions
646

647
```python
648
from fastmcp import FastMCP, Context
649
from fastmcp.utilities.types import Image
650
import json
651
import hashlib
652
from datetime import datetime
653
from typing import Any
654

655
def serialize_for_logging(obj: Any) -> str:
656
    """Serialize object for logging with proper handling of complex types."""
657
    if isinstance(obj, (str, int, float, bool, type(None))):
658
        return str(obj)
659
    elif isinstance(obj, (list, tuple)):
660
        return f"[{len(obj)} items]"
661
    elif isinstance(obj, dict):
662
        return f"{{keys: {list(obj.keys())[:5]}...}}"
663
    else:
664
        return f"<{type(obj).__name__}>"
665

666
def calculate_hash(data: str | bytes) -> str:
667
    """Calculate SHA-256 hash of data."""
668
    if isinstance(data, str):
669
        data = data.encode('utf-8')
670
    return hashlib.sha256(data).hexdigest()
671

672
def format_timestamp() -> str:
673
    """Get current timestamp in ISO format."""
674
    return datetime.utcnow().isoformat() + "Z"
675

676
mcp = FastMCP("Utility Demo Server")
677

678
@mcp.tool
679
async def process_data(
680
    data: dict,
681
    operation: str = "analyze",
682
    ctx: Context = None
683
) -> dict:
684
    """Process data with utility functions."""
685
    await ctx.info(f"Processing data: {serialize_for_logging(data)}")
686
    
687
    # Calculate data hash for integrity
688
    data_str = json.dumps(data, sort_keys=True)
689
    data_hash = calculate_hash(data_str)
690
    
691
    await ctx.info(f"Data hash: {data_hash}")
692
    
693
    # Process based on operation
694
    if operation == "analyze":
695
        result = {
696
            "analysis": f"Analyzed {len(data)} fields",
697
            "keys": list(data.keys())[:10],  # Limit for safety
698
            "field_types": {k: type(v).__name__ for k, v in data.items()}
699
        }
700
    elif operation == "transform":
701
        result = {
702
            "transformed": {k: str(v).upper() if isinstance(v, str) else v 
703
                          for k, v in data.items()},
704
            "transformation": "uppercase_strings"
705
        }
706
    else:
707
        result = {"error": f"Unknown operation: {operation}"}
708
    
709
    # Add metadata
710
    result.update({
711
        "data_hash": data_hash,
712
        "processed_at": format_timestamp(),
713
        "operation": operation,
714
        "status": "completed"
715
    })
716
    
717
    await ctx.info("Data processing completed")
718
    return result
719

720
@mcp.tool
721
def create_visualization(data: dict, chart_type: str = "bar") -> Image:
722
    """Create visualization from data."""
723
    import matplotlib.pyplot as plt
724
    import io
725
    
726
    # Extract data for plotting
727
    if isinstance(data, dict) and all(isinstance(v, (int, float)) for v in data.values()):
728
        keys = list(data.keys())
729
        values = list(data.values())
730
    else:
731
        # Fallback data
732
        keys = ["A", "B", "C", "D"]
733
        values = [1, 3, 2, 4]
734
    
735
    # Create chart
736
    plt.figure(figsize=(10, 6))
737
    
738
    if chart_type == "bar":
739
        plt.bar(keys, values)
740
    elif chart_type == "line":
741
        plt.plot(keys, values, marker='o')
742
    elif chart_type == "pie":
743
        plt.pie(values, labels=keys, autopct='%1.1f%%')
744
    else:
745
        plt.bar(keys, values)  # Default to bar
746
    
747
    plt.title(f"{chart_type.title()} Chart")
748
    plt.tight_layout()
749
    
750
    # Save to bytes
751
    buffer = io.BytesIO()
752
    plt.savefig(buffer, format='png', dpi=150)
753
    buffer.seek(0)
754
    
755
    return Image(buffer.read(), mime_type="image/png")
756

757
mcp.run()
758
```
759

760
## Type Aliases and Constants
761

762
```python { .api }
763
# Common type aliases
764
ToolFunction = Callable[..., Any]
765
ResourceFunction = Callable[..., str | bytes | dict]
766
PromptFunction = Callable[..., str | list[dict]]
767

768
# Transport types
769
TransportType = Literal["stdio", "sse", "http", "ws"]
770

771
# Log levels
772
LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
773

774
# Duplicate behavior options
775
DuplicateBehavior = Literal["error", "ignore", "replace"]
776

777
# Component types
778
ComponentFn = Callable[..., Any]
779
```