0
# Resources System
1

2
Resources expose read-only data sources with support for static resources and dynamic templates with URI parameters. The FastMCP resources system provides a clean way to expose data that LLMs can access for context and information retrieval.
3

4
## Capabilities
5

6
### Resource Classes
7

8
Base classes for creating and managing resources with different content types and access patterns.
9

10
```python { .api }
11
class Resource:
12
    def __init__(
13
        self,
14
        uri: str,
15
        name: str,
16
        description: str,
17
        mime_type: str | None = None
18
    ):
19
        """
20
        Base resource class for static resources.
21
        
22
        Parameters:
23
        - uri: Resource URI for client access
24
        - name: Human-readable resource name
25
        - description: Resource description for LLM understanding
26
        - mime_type: MIME type of resource content
27
        """
28

29
class FunctionResource(Resource):
30
    """Resource implementation for function-based resources."""
31

32
class TextResource:
33
    """Resource for text-based content."""
34

35
class BinaryResource:
36
    """Resource for binary content (images, files, etc.)."""
37

38
class FileResource:
39
    """Resource that serves file content from filesystem."""
40

41
class HttpResource:
42
    """Resource that proxies HTTP requests to external URLs."""
43

44
class DirectoryResource:
45
    """Resource that serves directory listings or file content."""
46
```
47

48
### Resource Templates
49

50
Dynamic resources that accept URI parameters for flexible data access.
51

52
```python { .api }
53
class ResourceTemplate:
54
    def __init__(
55
        self,
56
        uri_template: str,
57
        name: str,
58
        description: str,
59
        mime_type: str | None = None
60
    ):
61
        """
62
        Resource template for dynamic resources with URI parameters.
63
        
64
        Parameters:
65
        - uri_template: URI template with {parameters} placeholders
66
        - name: Human-readable template name
67
        - description: Template description for LLM understanding
68
        - mime_type: MIME type of resource content
69
        """
70
```
71

72
### Resource Manager
73

74
Manages resource registration, templates, and resolution within a FastMCP server.
75

76
```python { .api }
77
class ResourceManager:
78
    def add_resource(self, resource: Resource) -> None:
79
        """
80
        Add a static resource to the manager.
81
        
82
        Parameters:
83
        - resource: Resource instance to add
84
        """
85
    
86
    def add_template(self, template: ResourceTemplate) -> None:
87
        """
88
        Add a resource template for dynamic resources.
89
        
90
        Parameters:
91
        - template: ResourceTemplate instance to add
92
        """
93
    
94
    def get_resource(self, uri: str) -> Resource | None:
95
        """
96
        Get a resource by URI.
97
        
98
        Parameters:
99
        - uri: Resource URI to retrieve
100
        
101
        Returns:
102
        Resource instance or None if not found
103
        """
104
    
105
    def list_resources(self) -> list[Resource]:
106
        """
107
        List all registered static resources.
108
        
109
        Returns:
110
        List of all resource instances
111
        """
112
    
113
    def list_templates(self) -> list[ResourceTemplate]:
114
        """
115
        List all registered resource templates.
116
        
117
        Returns:
118
        List of all resource template instances
119
        """
120
```
121

122
## Usage Examples
123

124
### Basic Static Resources
125

126
```python
127
from fastmcp import FastMCP
128

129
mcp = FastMCP("Info Server")
130

131
@mcp.resource("config://version")
132
def get_version():
133
    """Get the current server version."""
134
    return "1.0.0"
135

136
@mcp.resource("config://settings")  
137
def get_settings():
138
    """Get server configuration settings."""
139
    return {
140
        "debug": True,
141
        "max_connections": 100,
142
        "timeout": 30
143
    }
144

145
@mcp.resource("info://status")
146
def get_status():
147
    """Get current server status."""
148
    import psutil
149
    import time
150
    
151
    return {
152
        "uptime": time.time() - mcp.start_time,
153
        "memory_usage": psutil.virtual_memory().percent,
154
        "cpu_usage": psutil.cpu_percent(),
155
        "active_connections": len(mcp.connections)
156
    }
157
```
158

159
### Dynamic Resource Templates
160

161
```python
162
from fastmcp import FastMCP
163

164
mcp = FastMCP("Data Server")
165

166
@mcp.resource("users://{user_id}/profile")
167
def get_user_profile(user_id: int):
168
    """
169
    Get user profile information by ID.
170
    
171
    Parameters:
172
    - user_id: Unique user identifier
173
    
174
    Returns:
175
    User profile data as JSON
176
    """
177
    # This would typically query a database
178
    users_db = {
179
        1: {"name": "Alice", "email": "alice@example.com", "status": "active"},
180
        2: {"name": "Bob", "email": "bob@example.com", "status": "inactive"},
181
        3: {"name": "Charlie", "email": "charlie@example.com", "status": "active"}
182
    }
183
    
184
    user = users_db.get(user_id)
185
    if not user:
186
        raise ValueError(f"User {user_id} not found")
187
    
188
    return {
189
        "user_id": user_id,
190
        **user,
191
        "last_updated": "2024-01-01T00:00:00Z"
192
    }
193

194
@mcp.resource("data://{dataset}/{table}")
195
def get_table_data(dataset: str, table: str):
196
    """
197
    Get data from a specific dataset table.
198
    
199
    Parameters:
200
    - dataset: Dataset name
201
    - table: Table name within dataset
202
    
203
    Returns:
204
    Table data as JSON
205
    """
206
    # Validate dataset and table
207
    valid_datasets = ["sales", "marketing", "finance"]
208
    if dataset not in valid_datasets:
209
        raise ValueError(f"Invalid dataset: {dataset}")
210
    
211
    # Mock data structure
212
    data = {
213
        "sales": {
214
            "orders": [
215
                {"id": 1, "amount": 100.0, "date": "2024-01-01"},
216
                {"id": 2, "amount": 250.0, "date": "2024-01-02"}
217
            ],
218
            "customers": [
219
                {"id": 1, "name": "Customer A", "region": "North"},
220
                {"id": 2, "name": "Customer B", "region": "South"}
221
            ]
222
        },
223
        "marketing": {
224
            "campaigns": [
225
                {"id": 1, "name": "Q1 Campaign", "budget": 10000},
226
                {"id": 2, "name": "Q2 Campaign", "budget": 15000}
227
            ]
228
        }
229
    }
230
    
231
    if table not in data.get(dataset, {}):
232
        raise ValueError(f"Table {table} not found in dataset {dataset}")
233
    
234
    return {
235
        "dataset": dataset,
236
        "table": table,
237
        "data": data[dataset][table],
238
        "record_count": len(data[dataset][table])
239
    }
240

241
@mcp.resource("files://{path}")
242
def get_file_content(path: str):
243
    """
244
    Get file content by path (with security restrictions).
245
    
246
    Parameters:  
247
    - path: File path relative to allowed directory
248
    
249
    Returns:
250
    File content as text or binary data
251
    """
252
    import os
253
    from pathlib import Path
254
    
255
    # Security: restrict to allowed directory
256
    allowed_dir = Path("./data")
257
    safe_path = allowed_dir / path
258
    
259
    # Prevent directory traversal
260
    if not safe_path.resolve().is_relative_to(allowed_dir.resolve()):
261
        raise ValueError("Access denied: path outside allowed directory")
262
    
263
    if not safe_path.exists():
264
        raise ValueError(f"File not found: {path}")
265
    
266
    if safe_path.is_dir():
267
        # Return directory listing
268
        files = [f.name for f in safe_path.iterdir() if f.is_file()]
269
        dirs = [d.name for d in safe_path.iterdir() if d.is_dir()]
270
        return {
271
            "type": "directory",
272
            "path": path,
273
            "files": files,
274
            "directories": dirs
275
        }
276
    
277
    # Return file content
278
    try:
279
        with open(safe_path, 'r', encoding='utf-8') as f:
280
            content = f.read()
281
        return {
282
            "type": "text",
283
            "path": path,
284
            "content": content,
285
            "size": len(content)
286
        }
287
    except UnicodeDecodeError:
288
        # Binary file
289
        with open(safe_path, 'rb') as f:
290
            content = f.read()
291
        return {
292
            "type": "binary",
293
            "path": path,
294
            "size": len(content),
295
            "content_base64": content.hex()  # Or base64.b64encode(content).decode()
296
        }
297
```
298

299
### Complex Resource with Context
300

301
```python
302
from fastmcp import FastMCP, Context
303
import json
304

305
mcp = FastMCP("API Proxy Server")
306

307
@mcp.resource("api://{service}/{endpoint}")
308
async def proxy_api_request(service: str, endpoint: str, ctx: Context):
309
    """
310
    Proxy requests to external APIs with authentication and logging.
311
    
312
    Parameters:
313
    - service: Service name (maps to API base URL)
314
    - endpoint: API endpoint path
315
    - ctx: Execution context for HTTP requests and logging
316
    
317
    Returns:
318
    API response data
319
    """
320
    # Service URL mapping
321
    service_urls = {
322
        "weather": "https://api.weather.com/v1",
323
        "news": "https://api.news.com/v2", 
324
        "stocks": "https://api.stocks.com/v1"
325
    }
326
    
327
    if service not in service_urls:
328
        await ctx.error(f"Unknown service: {service}")
329
        raise ValueError(f"Service {service} not supported")
330
    
331
    base_url = service_urls[service]
332
    full_url = f"{base_url}/{endpoint}"
333
    
334
    await ctx.info(f"Proxying request to {service}: {endpoint}")
335
    
336
    try:
337
        # Make HTTP request with authentication
338
        headers = {
339
            "Authorization": f"Bearer {get_api_key(service)}",
340
            "User-Agent": "FastMCP-Proxy/1.0"
341
        }
342
        
343
        response = await ctx.http_request(
344
            method="GET",
345
            url=full_url,
346
            headers=headers
347
        )
348
        
349
        if response.status_code == 200:
350
            await ctx.info(f"Successfully fetched data from {service}")
351
            data = response.json()
352
            
353
            return {
354
                "service": service,
355
                "endpoint": endpoint,
356
                "data": data,
357
                "timestamp": "2024-01-01T00:00:00Z",
358
                "status": "success"
359
            }
360
        else:
361
            await ctx.error(f"API request failed: {response.status_code}")
362
            return {
363
                "service": service,
364
                "endpoint": endpoint,
365
                "error": f"HTTP {response.status_code}",
366
                "status": "error"
367
            }
368
            
369
    except Exception as e:
370
        await ctx.error(f"Request failed: {str(e)}")
371
        return {
372
            "service": service,
373
            "endpoint": endpoint,
374
            "error": str(e),
375
            "status": "error"
376
        }
377

378
def get_api_key(service: str) -> str:
379
    """Get API key for service (would load from secure config)."""
380
    api_keys = {
381
        "weather": "weather_api_key_here",
382
        "news": "news_api_key_here",
383
        "stocks": "stocks_api_key_here"
384
    }
385
    return api_keys.get(service, "")
386
```
387

388
### File-Based Resources
389

390
```python
391
from fastmcp import FastMCP
392
from fastmcp.utilities.types import Image, File
393
import mimetypes
394
from pathlib import Path
395

396
mcp = FastMCP("File Server")
397

398
@mcp.resource("docs://{filename}")
399
def get_document(filename: str):
400
    """
401
    Serve documentation files.
402
    
403
    Parameters:
404
    - filename: Name of documentation file
405
    
406
    Returns:
407
    Document content with appropriate MIME type
408
    """
409
    docs_dir = Path("./docs")
410
    file_path = docs_dir / filename
411
    
412
    # Security check
413
    if not file_path.resolve().is_relative_to(docs_dir.resolve()):
414
        raise ValueError("Access denied")
415
    
416
    if not file_path.exists():
417
        raise ValueError(f"Document not found: {filename}")
418
    
419
    mime_type, _ = mimetypes.guess_type(str(file_path))
420
    
421
    with open(file_path, 'r', encoding='utf-8') as f:
422
        content = f.read()
423
    
424
    return {
425
        "filename": filename,
426
        "content": content,
427
        "mime_type": mime_type or "text/plain",
428
        "size": len(content)
429
    }
430

431
@mcp.resource("images://{image_name}")
432
def get_image(image_name: str):
433
    """
434
    Serve image files as Image objects.
435
    
436
    Parameters:
437
    - image_name: Name of image file
438
    
439
    Returns:
440
    Image object with binary data
441
    """
442
    images_dir = Path("./images")
443
    image_path = images_dir / image_name
444
    
445
    if not image_path.exists():
446
        raise ValueError(f"Image not found: {image_name}")
447
    
448
    with open(image_path, 'rb') as f:
449
        image_data = f.read()
450
    
451
    mime_type, _ = mimetypes.guess_type(str(image_path))
452
    
453
    return Image(
454
        data=image_data,
455
        mime_type=mime_type or "application/octet-stream"
456
    )
457

458
@mcp.resource("downloads://{file_name}")
459
def get_download_file(file_name: str):
460
    """
461
    Serve downloadable files as File objects.
462
    
463
    Parameters:
464
    - file_name: Name of file to download
465
    
466
    Returns:
467
    File object with metadata
468
    """
469
    downloads_dir = Path("./downloads")
470
    file_path = downloads_dir / file_name
471
    
472
    if not file_path.exists():
473
        raise ValueError(f"File not found: {file_name}")
474
    
475
    mime_type, _ = mimetypes.guess_type(str(file_path))
476
    
477
    if mime_type and mime_type.startswith('text/'):
478
        # Text file
479
        with open(file_path, 'r', encoding='utf-8') as f:
480
            content = f.read()
481
    else:
482
        # Binary file
483
        with open(file_path, 'rb') as f:
484
            content = f.read()
485
    
486
    return File(
487
        data=content,
488
        name=file_name,
489
        mime_type=mime_type
490
    )
491
```
492

493
### Database Resources
494

495
```python
496
from fastmcp import FastMCP
497
import sqlite3
498
import json
499

500
mcp = FastMCP("Database Server")
501

502
@mcp.resource("db://{table}")
503
def get_table_contents(table: str):
504
    """
505
    Get contents of a database table.
506
    
507
    Parameters:
508
    - table: Name of database table
509
    
510
    Returns:
511
    Table data as JSON with metadata
512
    """
513
    # Whitelist allowed tables for security
514
    allowed_tables = ["users", "orders", "products", "categories"]
515
    if table not in allowed_tables:
516
        raise ValueError(f"Access to table '{table}' not allowed")
517
    
518
    conn = sqlite3.connect("database.db")
519
    conn.row_factory = sqlite3.Row  # Enable column access by name
520
    cursor = conn.cursor()
521
    
522
    try:
523
        # Get table schema
524
        cursor.execute(f"PRAGMA table_info({table})")
525
        columns = [row['name'] for row in cursor.fetchall()]
526
        
527
        # Get table data
528
        cursor.execute(f"SELECT * FROM {table} LIMIT 100")  # Limit for safety
529
        rows = cursor.fetchall()
530
        
531
        # Convert to list of dictionaries
532
        data = [dict(row) for row in rows]
533
        
534
        return {
535
            "table": table,
536
            "columns": columns,
537
            "row_count": len(data),
538
            "data": data,
539
            "query_timestamp": "2024-01-01T00:00:00Z"
540
        }
541
        
542
    except sqlite3.Error as e:
543
        raise ValueError(f"Database error: {str(e)}")
544
    finally:
545
        conn.close()
546

547
@mcp.resource("db://query/{query_name}")
548
def get_predefined_query(query_name: str):
549
    """
550
    Execute predefined database queries.
551
    
552
    Parameters:
553
    - query_name: Name of predefined query
554
    
555
    Returns:
556
    Query results as JSON
557
    """
558
    # Predefined queries for security
559
    queries = {
560
        "user_stats": "SELECT COUNT(*) as total_users, COUNT(CASE WHEN status='active' THEN 1 END) as active_users FROM users",
561
        "recent_orders": "SELECT * FROM orders WHERE created_at > datetime('now', '-7 days') ORDER BY created_at DESC",
562
        "top_products": "SELECT p.*, COUNT(o.product_id) as order_count FROM products p LEFT JOIN orders o ON p.id = o.product_id GROUP BY p.id ORDER BY order_count DESC LIMIT 10"
563
    }
564
    
565
    if query_name not in queries:
566
        raise ValueError(f"Query '{query_name}' not found")
567
    
568
    conn = sqlite3.connect("database.db")
569
    conn.row_factory = sqlite3.Row
570
    cursor = conn.cursor()
571
    
572
    try:
573
        cursor.execute(queries[query_name])
574
        rows = cursor.fetchall()
575
        data = [dict(row) for row in rows]
576
        
577
        return {
578
            "query_name": query_name,
579
            "sql": queries[query_name],
580
            "result_count": len(data),
581
            "data": data,
582
            "executed_at": "2024-01-01T00:00:00Z"
583
        }
584
        
585
    except sqlite3.Error as e:
586
        raise ValueError(f"Query error: {str(e)}")
587
    finally:
588
        conn.close()
589
```
590

591
## Resource Content Types
592

593
Resources can return various content types:
594

595
```python
596
# Text content
597
@mcp.resource("text://example")
598
def text_resource():
599
    return "Plain text content"
600

601
# JSON data
602
@mcp.resource("json://example") 
603
def json_resource():
604
    return {"key": "value", "numbers": [1, 2, 3]}
605

606
# Binary data as Image
607
@mcp.resource("image://example")
608
def image_resource():
609
    with open("example.png", "rb") as f:
610
        return Image(f.read(), mime_type="image/png")
611

612
# Binary data as File
613
@mcp.resource("file://example")
614
def file_resource():
615
    return File(
616
        data=b"binary content",
617
        name="example.bin",
618
        mime_type="application/octet-stream"
619
    )
620
```