0
# Context and Dependencies
1
2
Execution context providing capabilities like logging, LLM sampling, HTTP requests, and resource access to tools/resources/prompts. The Context system enables FastMCP components to interact with the broader MCP ecosystem and client capabilities.
3
4
## Capabilities
5
6
### Context Class
7
8
Main context class providing execution capabilities for tools, resources, and prompts.
9
10
```python { .api }
11
class Context:
12
async def info(self, message: str) -> None:
13
"""
14
Log an info message to the client.
15
16
Parameters:
17
- message: Info message to log
18
"""
19
20
async def error(self, message: str) -> None:
21
"""
22
Log an error message to the client.
23
24
Parameters:
25
- message: Error message to log
26
"""
27
28
async def debug(self, message: str) -> None:
29
"""
30
Log a debug message to the client.
31
32
Parameters:
33
- message: Debug message to log
34
"""
35
36
async def warning(self, message: str) -> None:
37
"""
38
Log a warning message to the client.
39
40
Parameters:
41
- message: Warning message to log
42
"""
43
```
44
45
### LLM Sampling
46
47
Request LLM completions from the connected client.
48
49
```python { .api }
50
async def sample(
51
self,
52
messages: list[dict],
53
params: dict | None = None
54
) -> SamplingResult:
55
"""
56
Request LLM completion from the client.
57
58
Parameters:
59
- messages: List of message objects for the LLM
60
- params: Optional sampling parameters (temperature, max_tokens, etc.)
61
62
Returns:
63
Sampling result with generated text and metadata
64
"""
65
```
66
67
### Resource Access
68
69
Read resources from the server within tool/resource/prompt execution.
70
71
```python { .api }
72
async def read_resource(self, uri: str) -> ResourceResult:
73
"""
74
Read a resource from the server.
75
76
Parameters:
77
- uri: Resource URI to read
78
79
Returns:
80
Resource content and metadata
81
"""
82
```
83
84
### HTTP Requests
85
86
Make HTTP requests to external services with authentication and headers.
87
88
```python { .api }
89
async def http_request(
90
self,
91
method: str,
92
url: str,
93
headers: dict | None = None,
94
data: Any | None = None,
95
json: dict | None = None,
96
params: dict | None = None,
97
timeout: float = 30.0
98
) -> HttpResponse:
99
"""
100
Make HTTP request to external service.
101
102
Parameters:
103
- method: HTTP method (GET, POST, PUT, DELETE, etc.)
104
- url: Target URL
105
- headers: Optional request headers
106
- data: Request body as bytes/string
107
- json: Request body as JSON object
108
- params: URL query parameters
109
- timeout: Request timeout in seconds
110
111
Returns:
112
HTTP response with status, headers, and content
113
"""
114
```
115
116
### Progress Reporting
117
118
Report progress to the client for long-running operations.
119
120
```python { .api }
121
async def report_progress(
122
self,
123
progress: int,
124
total: int | None = None
125
) -> None:
126
"""
127
Report progress to the client.
128
129
Parameters:
130
- progress: Current progress value
131
- total: Total progress value (optional)
132
"""
133
```
134
135
### Dependency Injection Functions
136
137
Helper functions for accessing context and request information within components.
138
139
```python { .api }
140
def get_context() -> Context:
141
"""
142
Get the current execution context.
143
144
Returns:
145
Current Context instance
146
"""
147
148
def get_http_request() -> HttpRequest:
149
"""
150
Get the current HTTP request object (for HTTP transport).
151
152
Returns:
153
Current HTTP request
154
"""
155
156
def get_http_headers() -> dict[str, str]:
157
"""
158
Get the current HTTP request headers (for HTTP transport).
159
160
Returns:
161
Dictionary of HTTP headers
162
"""
163
164
def get_access_token() -> AccessToken | None:
165
"""
166
Get the current access token (if authenticated).
167
168
Returns:
169
Access token or None if not authenticated
170
"""
171
```
172
173
### Access Token
174
175
Access token representation for authenticated requests.
176
177
```python { .api }
178
class AccessToken:
179
token: str
180
token_type: str = "bearer"
181
expires_at: datetime | None = None
182
scope: list[str] | None = None
183
```
184
185
## Usage Examples
186
187
### Basic Context Usage
188
189
```python
190
from fastmcp import FastMCP, Context
191
192
mcp = FastMCP("Context Demo Server")
193
194
@mcp.tool
195
async def analyze_data(data: str, ctx: Context) -> str:
196
"""Analyze data with context logging and progress."""
197
await ctx.info("Starting data analysis")
198
199
# Report initial progress
200
await ctx.report_progress(0, 100)
201
202
# Simulate processing steps
203
steps = ["validation", "parsing", "analysis", "formatting"]
204
205
for i, step in enumerate(steps):
206
await ctx.info(f"Performing {step}")
207
208
# Simulate work
209
import asyncio
210
await asyncio.sleep(0.1)
211
212
# Report progress
213
progress = int((i + 1) / len(steps) * 100)
214
await ctx.report_progress(progress, 100)
215
216
await ctx.info("Data analysis completed")
217
return f"Analysis complete: {len(data)} characters processed"
218
219
@mcp.tool
220
async def safe_operation(input_data: str, ctx: Context) -> str:
221
"""Perform operation with error handling and logging."""
222
try:
223
await ctx.info("Starting safe operation")
224
225
if not input_data:
226
await ctx.warning("Empty input data provided")
227
return "No data to process"
228
229
# Process data
230
result = input_data.upper().strip()
231
await ctx.info(f"Processed {len(input_data)} characters")
232
233
return result
234
235
except Exception as e:
236
await ctx.error(f"Operation failed: {str(e)}")
237
raise
238
```
239
240
### LLM Sampling with Context
241
242
```python
243
from fastmcp import FastMCP, Context
244
245
mcp = FastMCP("LLM Integration Server")
246
247
@mcp.tool
248
async def intelligent_summary(
249
text: str,
250
max_words: int = 100,
251
ctx: Context = None
252
) -> str:
253
"""Generate intelligent summary using client's LLM."""
254
await ctx.info(f"Generating summary of {len(text)} character text")
255
256
# Prepare messages for LLM
257
messages = [
258
{
259
"role": "system",
260
"content": f"You are a professional summarizer. Create concise summaries in exactly {max_words} words or less."
261
},
262
{
263
"role": "user",
264
"content": f"Please summarize the following text:\n\n{text}"
265
}
266
]
267
268
# Request completion from client's LLM
269
result = await ctx.sample(
270
messages=messages,
271
params={
272
"temperature": 0.3,
273
"max_tokens": max_words * 2 # Buffer for token estimation
274
}
275
)
276
277
await ctx.info("Summary generated successfully")
278
return result.text
279
280
@mcp.tool
281
async def code_explanation(
282
code: str,
283
language: str,
284
ctx: Context
285
) -> str:
286
"""Explain code using LLM sampling."""
287
await ctx.info(f"Explaining {language} code")
288
289
messages = [
290
{
291
"role": "system",
292
"content": f"You are a programming expert. Explain {language} code in clear, simple terms."
293
},
294
{
295
"role": "user",
296
"content": f"Explain this {language} code:\n\n```{language}\n{code}\n```"
297
}
298
]
299
300
explanation = await ctx.sample(messages)
301
await ctx.info("Code explanation generated")
302
303
return explanation.text
304
305
@mcp.tool
306
async def creative_content(
307
prompt: str,
308
style: str = "professional",
309
ctx: Context = None
310
) -> str:
311
"""Generate creative content with LLM."""
312
await ctx.info(f"Generating {style} content")
313
314
messages = [
315
{
316
"role": "system",
317
"content": f"You are a {style} content creator. Write engaging, high-quality content."
318
},
319
{
320
"role": "user",
321
"content": prompt
322
}
323
]
324
325
# Use sampling parameters for creativity
326
params = {
327
"temperature": 0.8 if style == "creative" else 0.5,
328
"max_tokens": 1000
329
}
330
331
content = await ctx.sample(messages, params)
332
await ctx.info("Content generated successfully")
333
334
return content.text
335
```
336
337
### HTTP Requests with Context
338
339
```python
340
from fastmcp import FastMCP, Context
341
import json
342
343
mcp = FastMCP("API Integration Server")
344
345
@mcp.tool
346
async def fetch_weather(city: str, ctx: Context) -> dict:
347
"""Fetch weather data from external API."""
348
await ctx.info(f"Fetching weather for {city}")
349
350
# Make HTTP request to weather API
351
response = await ctx.http_request(
352
method="GET",
353
url="https://api.weather.com/v1/current",
354
params={"city": city, "units": "metric"},
355
headers={
356
"Authorization": "Bearer weather-api-key",
357
"User-Agent": "FastMCP-Weather/1.0"
358
},
359
timeout=10.0
360
)
361
362
if response.status_code == 200:
363
weather_data = response.json()
364
await ctx.info("Weather data retrieved successfully")
365
366
return {
367
"city": city,
368
"temperature": weather_data.get("temp"),
369
"condition": weather_data.get("condition"),
370
"humidity": weather_data.get("humidity"),
371
"source": "weather.com"
372
}
373
else:
374
await ctx.error(f"Weather API error: {response.status_code}")
375
return {"error": "Failed to fetch weather data"}
376
377
@mcp.tool
378
async def post_data(
379
endpoint: str,
380
data: dict,
381
ctx: Context
382
) -> dict:
383
"""Post data to external endpoint."""
384
await ctx.info(f"Posting data to {endpoint}")
385
386
response = await ctx.http_request(
387
method="POST",
388
url=endpoint,
389
json=data,
390
headers={
391
"Content-Type": "application/json",
392
"Authorization": "Bearer api-token"
393
}
394
)
395
396
if response.status_code in [200, 201]:
397
await ctx.info("Data posted successfully")
398
return {
399
"success": True,
400
"response": response.json(),
401
"status_code": response.status_code
402
}
403
else:
404
await ctx.error(f"POST failed: {response.status_code}")
405
return {
406
"success": False,
407
"error": response.text,
408
"status_code": response.status_code
409
}
410
411
@mcp.tool
412
async def aggregate_apis(
413
endpoints: list[str],
414
ctx: Context
415
) -> dict:
416
"""Aggregate data from multiple API endpoints."""
417
await ctx.info(f"Aggregating data from {len(endpoints)} endpoints")
418
419
results = {}
420
total_endpoints = len(endpoints)
421
422
for i, endpoint in enumerate(endpoints):
423
await ctx.info(f"Fetching from endpoint {i+1}/{total_endpoints}")
424
await ctx.report_progress(i, total_endpoints)
425
426
try:
427
response = await ctx.http_request("GET", endpoint, timeout=5.0)
428
429
if response.status_code == 200:
430
results[endpoint] = {
431
"success": True,
432
"data": response.json()
433
}
434
else:
435
results[endpoint] = {
436
"success": False,
437
"error": f"HTTP {response.status_code}"
438
}
439
440
except Exception as e:
441
await ctx.warning(f"Failed to fetch from {endpoint}: {str(e)}")
442
results[endpoint] = {
443
"success": False,
444
"error": str(e)
445
}
446
447
await ctx.report_progress(total_endpoints, total_endpoints)
448
await ctx.info("API aggregation completed")
449
450
# Summary statistics
451
successful = sum(1 for r in results.values() if r["success"])
452
453
return {
454
"summary": {
455
"total_endpoints": total_endpoints,
456
"successful": successful,
457
"failed": total_endpoints - successful
458
},
459
"results": results
460
}
461
```
462
463
### Resource Access with Context
464
465
```python
466
from fastmcp import FastMCP, Context
467
468
mcp = FastMCP("Resource Integration Server")
469
470
@mcp.tool
471
async def process_config(setting_name: str, ctx: Context) -> dict:
472
"""Process configuration using resource access."""
473
await ctx.info(f"Processing configuration: {setting_name}")
474
475
# Read configuration resource
476
config_resource = await ctx.read_resource("config://settings")
477
config_data = json.loads(config_resource.content)
478
479
if setting_name not in config_data:
480
await ctx.warning(f"Setting '{setting_name}' not found in configuration")
481
return {"error": f"Setting '{setting_name}' not found"}
482
483
setting_value = config_data[setting_name]
484
await ctx.info(f"Retrieved setting: {setting_name} = {setting_value}")
485
486
return {
487
"setting": setting_name,
488
"value": setting_value,
489
"type": type(setting_value).__name__
490
}
491
492
@mcp.tool
493
async def compile_report(report_type: str, ctx: Context) -> str:
494
"""Compile report using multiple resources."""
495
await ctx.info(f"Compiling {report_type} report")
496
497
# Read multiple resources
498
resources_to_read = [
499
"data://users/summary",
500
"data://orders/recent",
501
"config://report_settings"
502
]
503
504
report_data = {}
505
506
for i, resource_uri in enumerate(resources_to_read):
507
await ctx.info(f"Reading resource: {resource_uri}")
508
await ctx.report_progress(i, len(resources_to_read))
509
510
try:
511
resource = await ctx.read_resource(resource_uri)
512
report_data[resource_uri] = json.loads(resource.content)
513
except Exception as e:
514
await ctx.warning(f"Failed to read {resource_uri}: {str(e)}")
515
report_data[resource_uri] = {"error": str(e)}
516
517
await ctx.report_progress(len(resources_to_read), len(resources_to_read))
518
519
# Compile report
520
report = f"# {report_type.title()} Report\n\n"
521
522
for uri, data in report_data.items():
523
if "error" in data:
524
report += f"## {uri}\nError: {data['error']}\n\n"
525
else:
526
report += f"## {uri}\n{json.dumps(data, indent=2)}\n\n"
527
528
await ctx.info("Report compilation completed")
529
return report
530
```
531
532
### Authentication and Dependency Access
533
534
```python
535
from fastmcp import FastMCP, Context
536
from fastmcp.server.dependencies import get_access_token, get_http_headers
537
538
mcp = FastMCP("Secure Server")
539
540
@mcp.tool
541
async def secure_operation(data: str, ctx: Context) -> dict:
542
"""Perform secure operation with authentication info."""
543
await ctx.info("Starting secure operation")
544
545
# Get access token
546
token = get_access_token()
547
if not token:
548
await ctx.error("No access token available")
549
return {"error": "Authentication required"}
550
551
await ctx.info(f"Authenticated with token type: {token.token_type}")
552
553
# Get HTTP headers if available
554
try:
555
headers = get_http_headers()
556
user_agent = headers.get("user-agent", "unknown")
557
await ctx.info(f"Request from: {user_agent}")
558
except:
559
await ctx.debug("No HTTP headers available (not HTTP transport)")
560
561
# Process data securely
562
result = {
563
"processed_data": data.upper(),
564
"user_info": {
565
"token_type": token.token_type,
566
"has_scope": bool(token.scope),
567
"expires_at": token.expires_at.isoformat() if token.expires_at else None
568
},
569
"timestamp": "2024-01-01T00:00:00Z"
570
}
571
572
await ctx.info("Secure operation completed")
573
return result
574
575
@mcp.tool
576
async def user_specific_action(action: str, ctx: Context) -> str:
577
"""Perform action based on user authentication."""
578
token = get_access_token()
579
580
if not token:
581
await ctx.error("Authentication required")
582
return "Error: Please authenticate to perform this action"
583
584
# Check token scopes
585
required_scope = f"{action}_access"
586
if token.scope and required_scope not in token.scope:
587
await ctx.warning(f"Insufficient permissions for action: {action}")
588
return f"Error: Missing required scope '{required_scope}'"
589
590
await ctx.info(f"Performing {action} for authenticated user")
591
592
# Perform the action
593
return f"Successfully performed {action} with proper authentication"
594
```
595
596
## Result Types
597
598
```python { .api }
599
class SamplingResult:
600
"""Result from LLM sampling request."""
601
text: str
602
finish_reason: str | None
603
usage: dict | None
604
605
class HttpResponse:
606
"""Result from HTTP request."""
607
status_code: int
608
headers: dict[str, str]
609
text: str
610
content: bytes
611
612
def json(self) -> dict:
613
"""Parse response as JSON."""
614
```