Spec RegistrySpec Registry

Help your agents use open-source better. Learn more.

Find usage specs for your project’s dependencies

>

pypi-fastmcp

Describes: pypipypi/fastmcp

Description
The fast, Pythonic way to build MCP servers and clients with minimal boilerplate code.
Author
tessl
Last updated

How to use

npx @tessl/cli registry install tessl/pypi-fastmcp@2.12.0

resources.md docs/

1
# Resources System
2
3
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.
4
5
## Capabilities
6
7
### Resource Classes
8
9
Base classes for creating and managing resources with different content types and access patterns.
10
11
```python { .api }
12
class Resource:
13
def __init__(
14
self,
15
uri: str,
16
name: str,
17
description: str,
18
mime_type: str | None = None
19
):
20
"""
21
Base resource class for static resources.
22
23
Parameters:
24
- uri: Resource URI for client access
25
- name: Human-readable resource name
26
- description: Resource description for LLM understanding
27
- mime_type: MIME type of resource content
28
"""
29
30
class FunctionResource(Resource):
31
"""Resource implementation for function-based resources."""
32
33
class TextResource:
34
"""Resource for text-based content."""
35
36
class BinaryResource:
37
"""Resource for binary content (images, files, etc.)."""
38
39
class FileResource:
40
"""Resource that serves file content from filesystem."""
41
42
class HttpResource:
43
"""Resource that proxies HTTP requests to external URLs."""
44
45
class DirectoryResource:
46
"""Resource that serves directory listings or file content."""
47
```
48
49
### Resource Templates
50
51
Dynamic resources that accept URI parameters for flexible data access.
52
53
```python { .api }
54
class ResourceTemplate:
55
def __init__(
56
self,
57
uri_template: str,
58
name: str,
59
description: str,
60
mime_type: str | None = None
61
):
62
"""
63
Resource template for dynamic resources with URI parameters.
64
65
Parameters:
66
- uri_template: URI template with {parameters} placeholders
67
- name: Human-readable template name
68
- description: Template description for LLM understanding
69
- mime_type: MIME type of resource content
70
"""
71
```
72
73
### Resource Manager
74
75
Manages resource registration, templates, and resolution within a FastMCP server.
76
77
```python { .api }
78
class ResourceManager:
79
def add_resource(self, resource: Resource) -> None:
80
"""
81
Add a static resource to the manager.
82
83
Parameters:
84
- resource: Resource instance to add
85
"""
86
87
def add_template(self, template: ResourceTemplate) -> None:
88
"""
89
Add a resource template for dynamic resources.
90
91
Parameters:
92
- template: ResourceTemplate instance to add
93
"""
94
95
def get_resource(self, uri: str) -> Resource | None:
96
"""
97
Get a resource by URI.
98
99
Parameters:
100
- uri: Resource URI to retrieve
101
102
Returns:
103
Resource instance or None if not found
104
"""
105
106
def list_resources(self) -> list[Resource]:
107
"""
108
List all registered static resources.
109
110
Returns:
111
List of all resource instances
112
"""
113
114
def list_templates(self) -> list[ResourceTemplate]:
115
"""
116
List all registered resource templates.
117
118
Returns:
119
List of all resource template instances
120
"""
121
```
122
123
## Usage Examples
124
125
### Basic Static Resources
126
127
```python
128
from fastmcp import FastMCP
129
130
mcp = FastMCP("Info Server")
131
132
@mcp.resource("config://version")
133
def get_version():
134
"""Get the current server version."""
135
return "1.0.0"
136
137
@mcp.resource("config://settings")
138
def get_settings():
139
"""Get server configuration settings."""
140
return {
141
"debug": True,
142
"max_connections": 100,
143
"timeout": 30
144
}
145
146
@mcp.resource("info://status")
147
def get_status():
148
"""Get current server status."""
149
import psutil
150
import time
151
152
return {
153
"uptime": time.time() - mcp.start_time,
154
"memory_usage": psutil.virtual_memory().percent,
155
"cpu_usage": psutil.cpu_percent(),
156
"active_connections": len(mcp.connections)
157
}
158
```
159
160
### Dynamic Resource Templates
161
162
```python
163
from fastmcp import FastMCP
164
165
mcp = FastMCP("Data Server")
166
167
@mcp.resource("users://{user_id}/profile")
168
def get_user_profile(user_id: int):
169
"""
170
Get user profile information by ID.
171
172
Parameters:
173
- user_id: Unique user identifier
174
175
Returns:
176
User profile data as JSON
177
"""
178
# This would typically query a database
179
users_db = {
180
1: {"name": "Alice", "email": "alice@example.com", "status": "active"},
181
2: {"name": "Bob", "email": "bob@example.com", "status": "inactive"},
182
3: {"name": "Charlie", "email": "charlie@example.com", "status": "active"}
183
}
184
185
user = users_db.get(user_id)
186
if not user:
187
raise ValueError(f"User {user_id} not found")
188
189
return {
190
"user_id": user_id,
191
**user,
192
"last_updated": "2024-01-01T00:00:00Z"
193
}
194
195
@mcp.resource("data://{dataset}/{table}")
196
def get_table_data(dataset: str, table: str):
197
"""
198
Get data from a specific dataset table.
199
200
Parameters:
201
- dataset: Dataset name
202
- table: Table name within dataset
203
204
Returns:
205
Table data as JSON
206
"""
207
# Validate dataset and table
208
valid_datasets = ["sales", "marketing", "finance"]
209
if dataset not in valid_datasets:
210
raise ValueError(f"Invalid dataset: {dataset}")
211
212
# Mock data structure
213
data = {
214
"sales": {
215
"orders": [
216
{"id": 1, "amount": 100.0, "date": "2024-01-01"},
217
{"id": 2, "amount": 250.0, "date": "2024-01-02"}
218
],
219
"customers": [
220
{"id": 1, "name": "Customer A", "region": "North"},
221
{"id": 2, "name": "Customer B", "region": "South"}
222
]
223
},
224
"marketing": {
225
"campaigns": [
226
{"id": 1, "name": "Q1 Campaign", "budget": 10000},
227
{"id": 2, "name": "Q2 Campaign", "budget": 15000}
228
]
229
}
230
}
231
232
if table not in data.get(dataset, {}):
233
raise ValueError(f"Table {table} not found in dataset {dataset}")
234
235
return {
236
"dataset": dataset,
237
"table": table,
238
"data": data[dataset][table],
239
"record_count": len(data[dataset][table])
240
}
241
242
@mcp.resource("files://{path}")
243
def get_file_content(path: str):
244
"""
245
Get file content by path (with security restrictions).
246
247
Parameters:
248
- path: File path relative to allowed directory
249
250
Returns:
251
File content as text or binary data
252
"""
253
import os
254
from pathlib import Path
255
256
# Security: restrict to allowed directory
257
allowed_dir = Path("./data")
258
safe_path = allowed_dir / path
259
260
# Prevent directory traversal
261
if not safe_path.resolve().is_relative_to(allowed_dir.resolve()):
262
raise ValueError("Access denied: path outside allowed directory")
263
264
if not safe_path.exists():
265
raise ValueError(f"File not found: {path}")
266
267
if safe_path.is_dir():
268
# Return directory listing
269
files = [f.name for f in safe_path.iterdir() if f.is_file()]
270
dirs = [d.name for d in safe_path.iterdir() if d.is_dir()]
271
return {
272
"type": "directory",
273
"path": path,
274
"files": files,
275
"directories": dirs
276
}
277
278
# Return file content
279
try:
280
with open(safe_path, 'r', encoding='utf-8') as f:
281
content = f.read()
282
return {
283
"type": "text",
284
"path": path,
285
"content": content,
286
"size": len(content)
287
}
288
except UnicodeDecodeError:
289
# Binary file
290
with open(safe_path, 'rb') as f:
291
content = f.read()
292
return {
293
"type": "binary",
294
"path": path,
295
"size": len(content),
296
"content_base64": content.hex() # Or base64.b64encode(content).decode()
297
}
298
```
299
300
### Complex Resource with Context
301
302
```python
303
from fastmcp import FastMCP, Context
304
import json
305
306
mcp = FastMCP("API Proxy Server")
307
308
@mcp.resource("api://{service}/{endpoint}")
309
async def proxy_api_request(service: str, endpoint: str, ctx: Context):
310
"""
311
Proxy requests to external APIs with authentication and logging.
312
313
Parameters:
314
- service: Service name (maps to API base URL)
315
- endpoint: API endpoint path
316
- ctx: Execution context for HTTP requests and logging
317
318
Returns:
319
API response data
320
"""
321
# Service URL mapping
322
service_urls = {
323
"weather": "https://api.weather.com/v1",
324
"news": "https://api.news.com/v2",
325
"stocks": "https://api.stocks.com/v1"
326
}
327
328
if service not in service_urls:
329
await ctx.error(f"Unknown service: {service}")
330
raise ValueError(f"Service {service} not supported")
331
332
base_url = service_urls[service]
333
full_url = f"{base_url}/{endpoint}"
334
335
await ctx.info(f"Proxying request to {service}: {endpoint}")
336
337
try:
338
# Make HTTP request with authentication
339
headers = {
340
"Authorization": f"Bearer {get_api_key(service)}",
341
"User-Agent": "FastMCP-Proxy/1.0"
342
}
343
344
response = await ctx.http_request(
345
method="GET",
346
url=full_url,
347
headers=headers
348
)
349
350
if response.status_code == 200:
351
await ctx.info(f"Successfully fetched data from {service}")
352
data = response.json()
353
354
return {
355
"service": service,
356
"endpoint": endpoint,
357
"data": data,
358
"timestamp": "2024-01-01T00:00:00Z",
359
"status": "success"
360
}
361
else:
362
await ctx.error(f"API request failed: {response.status_code}")
363
return {
364
"service": service,
365
"endpoint": endpoint,
366
"error": f"HTTP {response.status_code}",
367
"status": "error"
368
}
369
370
except Exception as e:
371
await ctx.error(f"Request failed: {str(e)}")
372
return {
373
"service": service,
374
"endpoint": endpoint,
375
"error": str(e),
376
"status": "error"
377
}
378
379
def get_api_key(service: str) -> str:
380
"""Get API key for service (would load from secure config)."""
381
api_keys = {
382
"weather": "weather_api_key_here",
383
"news": "news_api_key_here",
384
"stocks": "stocks_api_key_here"
385
}
386
return api_keys.get(service, "")
387
```
388
389
### File-Based Resources
390
391
```python
392
from fastmcp import FastMCP
393
from fastmcp.utilities.types import Image, File
394
import mimetypes
395
from pathlib import Path
396
397
mcp = FastMCP("File Server")
398
399
@mcp.resource("docs://{filename}")
400
def get_document(filename: str):
401
"""
402
Serve documentation files.
403
404
Parameters:
405
- filename: Name of documentation file
406
407
Returns:
408
Document content with appropriate MIME type
409
"""
410
docs_dir = Path("./docs")
411
file_path = docs_dir / filename
412
413
# Security check
414
if not file_path.resolve().is_relative_to(docs_dir.resolve()):
415
raise ValueError("Access denied")
416
417
if not file_path.exists():
418
raise ValueError(f"Document not found: {filename}")
419
420
mime_type, _ = mimetypes.guess_type(str(file_path))
421
422
with open(file_path, 'r', encoding='utf-8') as f:
423
content = f.read()
424
425
return {
426
"filename": filename,
427
"content": content,
428
"mime_type": mime_type or "text/plain",
429
"size": len(content)
430
}
431
432
@mcp.resource("images://{image_name}")
433
def get_image(image_name: str):
434
"""
435
Serve image files as Image objects.
436
437
Parameters:
438
- image_name: Name of image file
439
440
Returns:
441
Image object with binary data
442
"""
443
images_dir = Path("./images")
444
image_path = images_dir / image_name
445
446
if not image_path.exists():
447
raise ValueError(f"Image not found: {image_name}")
448
449
with open(image_path, 'rb') as f:
450
image_data = f.read()
451
452
mime_type, _ = mimetypes.guess_type(str(image_path))
453
454
return Image(
455
data=image_data,
456
mime_type=mime_type or "application/octet-stream"
457
)
458
459
@mcp.resource("downloads://{file_name}")
460
def get_download_file(file_name: str):
461
"""
462
Serve downloadable files as File objects.
463
464
Parameters:
465
- file_name: Name of file to download
466
467
Returns:
468
File object with metadata
469
"""
470
downloads_dir = Path("./downloads")
471
file_path = downloads_dir / file_name
472
473
if not file_path.exists():
474
raise ValueError(f"File not found: {file_name}")
475
476
mime_type, _ = mimetypes.guess_type(str(file_path))
477
478
if mime_type and mime_type.startswith('text/'):
479
# Text file
480
with open(file_path, 'r', encoding='utf-8') as f:
481
content = f.read()
482
else:
483
# Binary file
484
with open(file_path, 'rb') as f:
485
content = f.read()
486
487
return File(
488
data=content,
489
name=file_name,
490
mime_type=mime_type
491
)
492
```
493
494
### Database Resources
495
496
```python
497
from fastmcp import FastMCP
498
import sqlite3
499
import json
500
501
mcp = FastMCP("Database Server")
502
503
@mcp.resource("db://{table}")
504
def get_table_contents(table: str):
505
"""
506
Get contents of a database table.
507
508
Parameters:
509
- table: Name of database table
510
511
Returns:
512
Table data as JSON with metadata
513
"""
514
# Whitelist allowed tables for security
515
allowed_tables = ["users", "orders", "products", "categories"]
516
if table not in allowed_tables:
517
raise ValueError(f"Access to table '{table}' not allowed")
518
519
conn = sqlite3.connect("database.db")
520
conn.row_factory = sqlite3.Row # Enable column access by name
521
cursor = conn.cursor()
522
523
try:
524
# Get table schema
525
cursor.execute(f"PRAGMA table_info({table})")
526
columns = [row['name'] for row in cursor.fetchall()]
527
528
# Get table data
529
cursor.execute(f"SELECT * FROM {table} LIMIT 100") # Limit for safety
530
rows = cursor.fetchall()
531
532
# Convert to list of dictionaries
533
data = [dict(row) for row in rows]
534
535
return {
536
"table": table,
537
"columns": columns,
538
"row_count": len(data),
539
"data": data,
540
"query_timestamp": "2024-01-01T00:00:00Z"
541
}
542
543
except sqlite3.Error as e:
544
raise ValueError(f"Database error: {str(e)}")
545
finally:
546
conn.close()
547
548
@mcp.resource("db://query/{query_name}")
549
def get_predefined_query(query_name: str):
550
"""
551
Execute predefined database queries.
552
553
Parameters:
554
- query_name: Name of predefined query
555
556
Returns:
557
Query results as JSON
558
"""
559
# Predefined queries for security
560
queries = {
561
"user_stats": "SELECT COUNT(*) as total_users, COUNT(CASE WHEN status='active' THEN 1 END) as active_users FROM users",
562
"recent_orders": "SELECT * FROM orders WHERE created_at > datetime('now', '-7 days') ORDER BY created_at DESC",
563
"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"
564
}
565
566
if query_name not in queries:
567
raise ValueError(f"Query '{query_name}' not found")
568
569
conn = sqlite3.connect("database.db")
570
conn.row_factory = sqlite3.Row
571
cursor = conn.cursor()
572
573
try:
574
cursor.execute(queries[query_name])
575
rows = cursor.fetchall()
576
data = [dict(row) for row in rows]
577
578
return {
579
"query_name": query_name,
580
"sql": queries[query_name],
581
"result_count": len(data),
582
"data": data,
583
"executed_at": "2024-01-01T00:00:00Z"
584
}
585
586
except sqlite3.Error as e:
587
raise ValueError(f"Query error: {str(e)}")
588
finally:
589
conn.close()
590
```
591
592
## Resource Content Types
593
594
Resources can return various content types:
595
596
```python
597
# Text content
598
@mcp.resource("text://example")
599
def text_resource():
600
return "Plain text content"
601
602
# JSON data
603
@mcp.resource("json://example")
604
def json_resource():
605
return {"key": "value", "numbers": [1, 2, 3]}
606
607
# Binary data as Image
608
@mcp.resource("image://example")
609
def image_resource():
610
with open("example.png", "rb") as f:
611
return Image(f.read(), mime_type="image/png")
612
613
# Binary data as File
614
@mcp.resource("file://example")
615
def file_resource():
616
return File(
617
data=b"binary content",
618
name="example.bin",
619
mime_type="application/octet-stream"
620
)
621
```