0
# Utilities and Types
1

2
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.
3

4
## Capabilities
5

6
### Media and File Types
7

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

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

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

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

56
### Base Model Classes
57

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

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

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

81
### Component System
82

83
Base classes for creating reusable FastMCP components.
84

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

107
### Settings Management
108

109
Comprehensive settings system for configuring FastMCP behavior.
110

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

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

144
### OpenAPI Integration
145

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

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

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

195
### MCP Configuration
196

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

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

218
### Testing Utilities
219

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

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

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

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

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

236
### JSON Schema Utilities
237

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

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

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

265
### HTTP Utilities
266

267
Utilities for HTTP operations and request handling.
268

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

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

295
## Usage Examples
296

297
### Media Types Usage
298

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

304
mcp = FastMCP("Media Server")
305

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

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

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

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

373
### Settings Configuration
374

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

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

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

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

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

409
mcp.run()
410
```
411

412
### Component System Usage
413

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

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

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

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

476
mcp = FastMCP("Component Server")
477

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

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

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

509
### OpenAPI Integration
510

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

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

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

573
mcp = FastMCP("OpenAPI Server")
574

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

597
mcp.run()
598
```
599

600
### Testing Utilities Usage
601

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

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

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

644
### Custom Utility Functions
645

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

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

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

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

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

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

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

756
mcp.run()
757
```
758

759
## Type Aliases and Constants
760

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

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

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

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

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