1
# Context and Dependencies
2

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

5
## Capabilities
6

7
### Context Class
8

9
Main context class providing execution capabilities for tools, resources, and prompts.
10

11
```python { .api }
12
class Context:
13
    async def info(self, message: str) -> None:
14
        """
15
        Log an info message to the client.
16
        
17
        Parameters:
18
        - message: Info message to log
19
        """
20
    
21
    async def error(self, message: str) -> None:
22
        """
23
        Log an error message to the client.
24
        
25
        Parameters:
26
        - message: Error message to log
27
        """
28
    
29
    async def debug(self, message: str) -> None:
30
        """
31
        Log a debug message to the client.
32
        
33
        Parameters:
34
        - message: Debug message to log
35
        """
36
    
37
    async def warning(self, message: str) -> None:
38
        """
39
        Log a warning message to the client.
40
        
41
        Parameters:
42
        - message: Warning message to log
43
        """
44
```
45

46
### LLM Sampling
47

48
Request LLM completions from the connected client.
49

50
```python { .api }
51
async def sample(
52
    self,
53
    messages: list[dict],
54
    params: dict | None = None
55
) -> SamplingResult:
56
    """
57
    Request LLM completion from the client.
58
    
59
    Parameters:
60
    - messages: List of message objects for the LLM
61
    - params: Optional sampling parameters (temperature, max_tokens, etc.)
62
    
63
    Returns:
64
    Sampling result with generated text and metadata
65
    """
66
```
67

68
### Resource Access
69

70
Read resources from the server within tool/resource/prompt execution.
71

72
```python { .api }
73
async def read_resource(self, uri: str) -> ResourceResult:
74
    """
75
    Read a resource from the server.
76
    
77
    Parameters:
78
    - uri: Resource URI to read
79
    
80
    Returns:
81
    Resource content and metadata
82
    """
83
```
84

85
### HTTP Requests
86

87
Make HTTP requests to external services with authentication and headers.
88

89
```python { .api }
90
async def http_request(
91
    self,
92
    method: str,
93
    url: str,
94
    headers: dict | None = None,
95
    data: Any | None = None,
96
    json: dict | None = None,
97
    params: dict | None = None,
98
    timeout: float = 30.0
99
) -> HttpResponse:
100
    """
101
    Make HTTP request to external service.
102
    
103
    Parameters:
104
    - method: HTTP method (GET, POST, PUT, DELETE, etc.)
105
    - url: Target URL
106
    - headers: Optional request headers
107
    - data: Request body as bytes/string
108
    - json: Request body as JSON object
109
    - params: URL query parameters
110
    - timeout: Request timeout in seconds
111
    
112
    Returns:
113
    HTTP response with status, headers, and content
114
    """
115
```
116

117
### Progress Reporting
118

119
Report progress to the client for long-running operations.
120

121
```python { .api }
122
async def report_progress(
123
    self,
124
    progress: int,
125
    total: int | None = None
126
) -> None:
127
    """
128
    Report progress to the client.
129
    
130
    Parameters:
131
    - progress: Current progress value
132
    - total: Total progress value (optional)
133
    """
134
```
135

136
### Dependency Injection Functions
137

138
Helper functions for accessing context and request information within components.
139

140
```python { .api }
141
def get_context() -> Context:
142
    """
143
    Get the current execution context.
144
    
145
    Returns:
146
    Current Context instance
147
    """
148

149
def get_http_request() -> HttpRequest:
150
    """
151
    Get the current HTTP request object (for HTTP transport).
152
    
153
    Returns:
154
    Current HTTP request
155
    """
156

157
def get_http_headers() -> dict[str, str]:
158
    """
159
    Get the current HTTP request headers (for HTTP transport).
160
    
161
    Returns:
162
    Dictionary of HTTP headers
163
    """
164

165
def get_access_token() -> AccessToken | None:
166
    """
167
    Get the current access token (if authenticated).
168
    
169
    Returns:
170
    Access token or None if not authenticated
171
    """
172
```
173

174
### Access Token
175

176
Access token representation for authenticated requests.
177

178
```python { .api }
179
class AccessToken:
180
    token: str
181
    token_type: str = "bearer"
182
    expires_at: datetime | None = None
183
    scope: list[str] | None = None
184
```
185

186
## Usage Examples
187

188
### Basic Context Usage
189

190
```python
191
from fastmcp import FastMCP, Context
192

193
mcp = FastMCP("Context Demo Server")
194

195
@mcp.tool
196
async def analyze_data(data: str, ctx: Context) -> str:
197
    """Analyze data with context logging and progress."""
198
    await ctx.info("Starting data analysis")
199
    
200
    # Report initial progress
201
    await ctx.report_progress(0, 100)
202
    
203
    # Simulate processing steps
204
    steps = ["validation", "parsing", "analysis", "formatting"]
205
    
206
    for i, step in enumerate(steps):
207
        await ctx.info(f"Performing {step}")
208
        
209
        # Simulate work
210
        import asyncio
211
        await asyncio.sleep(0.1)
212
        
213
        # Report progress
214
        progress = int((i + 1) / len(steps) * 100)
215
        await ctx.report_progress(progress, 100)
216
    
217
    await ctx.info("Data analysis completed")
218
    return f"Analysis complete: {len(data)} characters processed"
219

220
@mcp.tool
221
async def safe_operation(input_data: str, ctx: Context) -> str:
222
    """Perform operation with error handling and logging."""
223
    try:
224
        await ctx.info("Starting safe operation")
225
        
226
        if not input_data:
227
            await ctx.warning("Empty input data provided")
228
            return "No data to process"
229
        
230
        # Process data
231
        result = input_data.upper().strip()
232
        await ctx.info(f"Processed {len(input_data)} characters")
233
        
234
        return result
235
        
236
    except Exception as e:
237
        await ctx.error(f"Operation failed: {str(e)}")
238
        raise
239
```
240

241
### LLM Sampling with Context
242

243
```python
244
from fastmcp import FastMCP, Context
245

246
mcp = FastMCP("LLM Integration Server")
247

248
@mcp.tool
249
async def intelligent_summary(
250
    text: str,
251
    max_words: int = 100,
252
    ctx: Context = None
253
) -> str:
254
    """Generate intelligent summary using client's LLM."""
255
    await ctx.info(f"Generating summary of {len(text)} character text")
256
    
257
    # Prepare messages for LLM
258
    messages = [
259
        {
260
            "role": "system",
261
            "content": f"You are a professional summarizer. Create concise summaries in exactly {max_words} words or less."
262
        },
263
        {
264
            "role": "user", 
265
            "content": f"Please summarize the following text:\n\n{text}"
266
        }
267
    ]
268
    
269
    # Request completion from client's LLM
270
    result = await ctx.sample(
271
        messages=messages,
272
        params={
273
            "temperature": 0.3,
274
            "max_tokens": max_words * 2  # Buffer for token estimation
275
        }
276
    )
277
    
278
    await ctx.info("Summary generated successfully")
279
    return result.text
280

281
@mcp.tool
282
async def code_explanation(
283
    code: str,
284
    language: str,
285
    ctx: Context
286
) -> str:
287
    """Explain code using LLM sampling."""
288
    await ctx.info(f"Explaining {language} code")
289
    
290
    messages = [
291
        {
292
            "role": "system",
293
            "content": f"You are a programming expert. Explain {language} code in clear, simple terms."
294
        },
295
        {
296
            "role": "user",
297
            "content": f"Explain this {language} code:\n\n```{language}\n{code}\n```"
298
        }
299
    ]
300
    
301
    explanation = await ctx.sample(messages)
302
    await ctx.info("Code explanation generated")
303
    
304
    return explanation.text
305

306
@mcp.tool
307
async def creative_content(
308
    prompt: str,
309
    style: str = "professional",
310
    ctx: Context = None
311
) -> str:
312
    """Generate creative content with LLM."""
313
    await ctx.info(f"Generating {style} content")
314
    
315
    messages = [
316
        {
317
            "role": "system",
318
            "content": f"You are a {style} content creator. Write engaging, high-quality content."
319
        },
320
        {
321
            "role": "user",
322
            "content": prompt
323
        }
324
    ]
325
    
326
    # Use sampling parameters for creativity
327
    params = {
328
        "temperature": 0.8 if style == "creative" else 0.5,
329
        "max_tokens": 1000
330
    }
331
    
332
    content = await ctx.sample(messages, params)
333
    await ctx.info("Content generated successfully")
334
    
335
    return content.text
336
```
337

338
### HTTP Requests with Context
339

340
```python
341
from fastmcp import FastMCP, Context
342
import json
343

344
mcp = FastMCP("API Integration Server")
345

346
@mcp.tool
347
async def fetch_weather(city: str, ctx: Context) -> dict:
348
    """Fetch weather data from external API."""
349
    await ctx.info(f"Fetching weather for {city}")
350
    
351
    # Make HTTP request to weather API
352
    response = await ctx.http_request(
353
        method="GET",
354
        url="https://api.weather.com/v1/current",
355
        params={"city": city, "units": "metric"},
356
        headers={
357
            "Authorization": "Bearer weather-api-key",
358
            "User-Agent": "FastMCP-Weather/1.0"
359
        },
360
        timeout=10.0
361
    )
362
    
363
    if response.status_code == 200:
364
        weather_data = response.json()
365
        await ctx.info("Weather data retrieved successfully")
366
        
367
        return {
368
            "city": city,
369
            "temperature": weather_data.get("temp"),
370
            "condition": weather_data.get("condition"),
371
            "humidity": weather_data.get("humidity"),
372
            "source": "weather.com"
373
        }
374
    else:
375
        await ctx.error(f"Weather API error: {response.status_code}")
376
        return {"error": "Failed to fetch weather data"}
377

378
@mcp.tool
379
async def post_data(
380
    endpoint: str,
381
    data: dict,
382
    ctx: Context
383
) -> dict:
384
    """Post data to external endpoint."""
385
    await ctx.info(f"Posting data to {endpoint}")
386
    
387
    response = await ctx.http_request(
388
        method="POST",
389
        url=endpoint,
390
        json=data,
391
        headers={
392
            "Content-Type": "application/json",
393
            "Authorization": "Bearer api-token"
394
        }
395
    )
396
    
397
    if response.status_code in [200, 201]:
398
        await ctx.info("Data posted successfully")
399
        return {
400
            "success": True,
401
            "response": response.json(),
402
            "status_code": response.status_code
403
        }
404
    else:
405
        await ctx.error(f"POST failed: {response.status_code}")
406
        return {
407
            "success": False,
408
            "error": response.text,
409
            "status_code": response.status_code
410
        }
411

412
@mcp.tool
413
async def aggregate_apis(
414
    endpoints: list[str],
415
    ctx: Context
416
) -> dict:
417
    """Aggregate data from multiple API endpoints."""
418
    await ctx.info(f"Aggregating data from {len(endpoints)} endpoints")
419
    
420
    results = {}
421
    total_endpoints = len(endpoints)
422
    
423
    for i, endpoint in enumerate(endpoints):
424
        await ctx.info(f"Fetching from endpoint {i+1}/{total_endpoints}")
425
        await ctx.report_progress(i, total_endpoints)
426
        
427
        try:
428
            response = await ctx.http_request("GET", endpoint, timeout=5.0)
429
            
430
            if response.status_code == 200:
431
                results[endpoint] = {
432
                    "success": True,
433
                    "data": response.json()
434
                }
435
            else:
436
                results[endpoint] = {
437
                    "success": False,
438
                    "error": f"HTTP {response.status_code}"
439
                }
440
                
441
        except Exception as e:
442
            await ctx.warning(f"Failed to fetch from {endpoint}: {str(e)}")
443
            results[endpoint] = {
444
                "success": False,
445
                "error": str(e)
446
            }
447
    
448
    await ctx.report_progress(total_endpoints, total_endpoints)
449
    await ctx.info("API aggregation completed")
450
    
451
    # Summary statistics
452
    successful = sum(1 for r in results.values() if r["success"])
453
    
454
    return {
455
        "summary": {
456
            "total_endpoints": total_endpoints,
457
            "successful": successful,
458
            "failed": total_endpoints - successful
459
        },
460
        "results": results
461
    }
462
```
463

464
### Resource Access with Context
465

466
```python
467
from fastmcp import FastMCP, Context
468

469
mcp = FastMCP("Resource Integration Server")
470

471
@mcp.tool
472
async def process_config(setting_name: str, ctx: Context) -> dict:
473
    """Process configuration using resource access."""
474
    await ctx.info(f"Processing configuration: {setting_name}")
475
    
476
    # Read configuration resource
477
    config_resource = await ctx.read_resource("config://settings")
478
    config_data = json.loads(config_resource.content)
479
    
480
    if setting_name not in config_data:
481
        await ctx.warning(f"Setting '{setting_name}' not found in configuration")
482
        return {"error": f"Setting '{setting_name}' not found"}
483
    
484
    setting_value = config_data[setting_name]
485
    await ctx.info(f"Retrieved setting: {setting_name} = {setting_value}")
486
    
487
    return {
488
        "setting": setting_name,
489
        "value": setting_value,
490
        "type": type(setting_value).__name__
491
    }
492

493
@mcp.tool
494
async def compile_report(report_type: str, ctx: Context) -> str:
495
    """Compile report using multiple resources."""
496
    await ctx.info(f"Compiling {report_type} report")
497
    
498
    # Read multiple resources
499
    resources_to_read = [
500
        "data://users/summary",
501
        "data://orders/recent", 
502
        "config://report_settings"
503
    ]
504
    
505
    report_data = {}
506
    
507
    for i, resource_uri in enumerate(resources_to_read):
508
        await ctx.info(f"Reading resource: {resource_uri}")
509
        await ctx.report_progress(i, len(resources_to_read))
510
        
511
        try:
512
            resource = await ctx.read_resource(resource_uri)  
513
            report_data[resource_uri] = json.loads(resource.content)
514
        except Exception as e:
515
            await ctx.warning(f"Failed to read {resource_uri}: {str(e)}")
516
            report_data[resource_uri] = {"error": str(e)}
517
    
518
    await ctx.report_progress(len(resources_to_read), len(resources_to_read))
519
    
520
    # Compile report
521
    report = f"# {report_type.title()} Report\n\n"
522
    
523
    for uri, data in report_data.items():
524
        if "error" in data:
525
            report += f"## {uri}\nError: {data['error']}\n\n"
526
        else:
527
            report += f"## {uri}\n{json.dumps(data, indent=2)}\n\n"
528
    
529
    await ctx.info("Report compilation completed")
530
    return report
531
```
532

533
### Authentication and Dependency Access
534

535
```python
536
from fastmcp import FastMCP, Context
537
from fastmcp.server.dependencies import get_access_token, get_http_headers
538

539
mcp = FastMCP("Secure Server")
540

541
@mcp.tool
542
async def secure_operation(data: str, ctx: Context) -> dict:
543
    """Perform secure operation with authentication info."""
544
    await ctx.info("Starting secure operation")
545
    
546
    # Get access token
547
    token = get_access_token()
548
    if not token:
549
        await ctx.error("No access token available")
550
        return {"error": "Authentication required"}
551
    
552
    await ctx.info(f"Authenticated with token type: {token.token_type}")
553
    
554
    # Get HTTP headers if available
555
    try:
556
        headers = get_http_headers()
557
        user_agent = headers.get("user-agent", "unknown")
558
        await ctx.info(f"Request from: {user_agent}")
559
    except:
560
        await ctx.debug("No HTTP headers available (not HTTP transport)")
561
    
562
    # Process data securely
563
    result = {
564
        "processed_data": data.upper(),
565
        "user_info": {
566
            "token_type": token.token_type,
567
            "has_scope": bool(token.scope),
568
            "expires_at": token.expires_at.isoformat() if token.expires_at else None
569
        },
570
        "timestamp": "2024-01-01T00:00:00Z"
571
    }
572
    
573
    await ctx.info("Secure operation completed")
574
    return result
575

576
@mcp.tool
577
async def user_specific_action(action: str, ctx: Context) -> str:
578
    """Perform action based on user authentication."""
579
    token = get_access_token()
580
    
581
    if not token:
582
        await ctx.error("Authentication required")
583
        return "Error: Please authenticate to perform this action"
584
    
585
    # Check token scopes
586
    required_scope = f"{action}_access"
587
    if token.scope and required_scope not in token.scope:
588
        await ctx.warning(f"Insufficient permissions for action: {action}")
589
        return f"Error: Missing required scope '{required_scope}'"
590
    
591
    await ctx.info(f"Performing {action} for authenticated user")
592
    
593
    # Perform the action
594
    return f"Successfully performed {action} with proper authentication"
595
```
596

597
## Result Types
598

599
```python { .api }
600
class SamplingResult:
601
    """Result from LLM sampling request."""
602
    text: str
603
    finish_reason: str | None
604
    usage: dict | None
605

606
class HttpResponse:
607
    """Result from HTTP request."""
608
    status_code: int
609
    headers: dict[str, str]
610
    text: str  
611
    content: bytes
612
    
613
    def json(self) -> dict:
614
        """Parse response as JSON."""
615
```