0
# Event Handlers
1
2
Framework for building Lambda functions that handle HTTP events from API Gateway, Application Load Balancer, AppSync GraphQL, Bedrock Agents, and other AWS services with automatic request/response serialization, routing, and CORS support.
3
4
## Capabilities
5
6
### API Gateway REST Resolver
7
8
HTTP request resolver for API Gateway REST APIs with automatic routing, request validation, and response formatting.
9
10
```python { .api }
11
class APIGatewayRestResolver:
12
def __init__(
13
self,
14
cors: CORSConfig = None,
15
debug: bool = None,
16
serializer: Callable[[Dict], str] = None,
17
strip_prefixes: List[str] = None,
18
enable_validation: bool = False,
19
):
20
"""
21
Initialize API Gateway REST resolver.
22
23
Parameters:
24
- cors: CORS configuration for cross-origin requests
25
- debug: Enable debug mode for detailed error responses
26
- serializer: Custom JSON serializer function
27
- strip_prefixes: Path prefixes to strip from routes
28
- enable_validation: Enable request validation
29
"""
30
31
def get(self, rule: str, **kwargs) -> Callable:
32
"""
33
Register GET route handler.
34
35
Parameters:
36
- rule: URL route pattern (supports path parameters)
37
- **kwargs: Additional route configuration
38
39
Returns:
40
Decorator function for route handler
41
"""
42
43
def post(self, rule: str, **kwargs) -> Callable:
44
"""Register POST route handler"""
45
46
def put(self, rule: str, **kwargs) -> Callable:
47
"""Register PUT route handler"""
48
49
def patch(self, rule: str, **kwargs) -> Callable:
50
"""Register PATCH route handler"""
51
52
def delete(self, rule: str, **kwargs) -> Callable:
53
"""Register DELETE route handler"""
54
55
def head(self, rule: str, **kwargs) -> Callable:
56
"""Register HEAD route handler"""
57
58
def options(self, rule: str, **kwargs) -> Callable:
59
"""Register OPTIONS route handler"""
60
61
def route(
62
self,
63
rule: str,
64
method: Union[str, List[str]],
65
**kwargs,
66
) -> Callable:
67
"""
68
Register route handler for specific HTTP method(s).
69
70
Parameters:
71
- rule: URL route pattern
72
- method: HTTP method or list of methods
73
- **kwargs: Additional route configuration
74
"""
75
76
def resolve(self, event: Dict, context: LambdaContext) -> Dict:
77
"""
78
Resolve incoming Lambda event to registered route handler.
79
80
Parameters:
81
- event: API Gateway Lambda proxy integration event
82
- context: Lambda runtime context
83
84
Returns:
85
API Gateway response dictionary
86
"""
87
88
@property
89
def current_event(self) -> APIGatewayProxyEvent:
90
"""Get current request event during handler execution"""
91
92
@property
93
def lambda_context(self) -> LambdaContext:
94
"""Get Lambda context during handler execution"""
95
96
def append_context(self, **kwargs) -> None:
97
"""Append key-value pairs to request context"""
98
99
def clear_context(self) -> None:
100
"""Clear request context"""
101
102
def exception_handler(self, exception_type: Exception) -> Callable:
103
"""
104
Register exception handler for specific exception type.
105
106
Parameters:
107
- exception_type: Exception class to handle
108
109
Returns:
110
Decorator function for exception handler
111
"""
112
113
def not_found(self) -> Callable:
114
"""Register handler for 404 Not Found responses"""
115
116
def unauthorized(self) -> Callable:
117
"""Register handler for 401 Unauthorized responses"""
118
119
def include_router(self, router: "APIGatewayRestResolver", prefix: str = None) -> None:
120
"""
121
Include routes from another resolver.
122
123
Parameters:
124
- router: Another APIGatewayRestResolver instance
125
- prefix: Optional path prefix for included routes
126
"""
127
128
def enable_swagger(
129
self,
130
path: str = "/swagger",
131
persist_authorization: bool = True,
132
**kwargs,
133
) -> None:
134
"""
135
Enable Swagger UI documentation.
136
137
Parameters:
138
- path: Path to serve Swagger UI
139
- persist_authorization: Whether to persist auth across requests
140
- **kwargs: Additional Swagger configuration
141
"""
142
```
143
144
### API Gateway HTTP Resolver
145
146
HTTP request resolver optimized for API Gateway HTTP APIs (v2.0) with improved performance and simpler payload format.
147
148
```python { .api }
149
class APIGatewayHttpResolver:
150
def __init__(
151
self,
152
cors: CORSConfig = None,
153
debug: bool = None,
154
serializer: Callable[[Dict], str] = None,
155
strip_prefixes: List[str] = None,
156
enable_validation: bool = False,
157
):
158
"""
159
Initialize API Gateway HTTP resolver for v2.0 format.
160
161
Parameters:
162
- cors: CORS configuration
163
- debug: Enable debug mode
164
- serializer: Custom JSON serializer
165
- strip_prefixes: Path prefixes to strip
166
- enable_validation: Enable request validation
167
"""
168
169
# Inherits same route methods as APIGatewayRestResolver
170
def get(self, rule: str, **kwargs) -> Callable: ...
171
def post(self, rule: str, **kwargs) -> Callable: ...
172
def put(self, rule: str, **kwargs) -> Callable: ...
173
def delete(self, rule: str, **kwargs) -> Callable: ...
174
def route(self, rule: str, method: Union[str, List[str]], **kwargs) -> Callable: ...
175
176
def resolve(self, event: Dict, context: LambdaContext) -> Dict:
177
"""Resolve HTTP API v2.0 event to route handler"""
178
179
@property
180
def current_event(self) -> APIGatewayProxyEventV2:
181
"""Get current HTTP API v2.0 request event"""
182
```
183
184
### Application Load Balancer Resolver
185
186
HTTP request resolver for Application Load Balancer target group integration.
187
188
```python { .api }
189
class ALBResolver:
190
def __init__(
191
self,
192
cors: CORSConfig = None,
193
debug: bool = None,
194
serializer: Callable[[Dict], str] = None,
195
strip_prefixes: List[str] = None,
196
enable_validation: bool = False,
197
):
198
"""
199
Initialize Application Load Balancer resolver.
200
201
Parameters:
202
- cors: CORS configuration
203
- debug: Enable debug mode
204
- serializer: Custom JSON serializer
205
- strip_prefixes: Path prefixes to strip
206
- enable_validation: Enable request validation
207
"""
208
209
# Same routing interface as API Gateway resolvers
210
def get(self, rule: str, **kwargs) -> Callable: ...
211
def post(self, rule: str, **kwargs) -> Callable: ...
212
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
213
214
@property
215
def current_event(self) -> ALBEvent:
216
"""Get current ALB request event"""
217
```
218
219
### AppSync Resolver
220
221
GraphQL resolver for AWS AppSync with support for direct Lambda resolvers and batch resolvers.
222
223
```python { .api }
224
class AppSyncResolver:
225
def __init__(self, debug: bool = None):
226
"""
227
Initialize AppSync GraphQL resolver.
228
229
Parameters:
230
- debug: Enable debug mode for detailed error responses
231
"""
232
233
def resolver(
234
self,
235
type_name: str = "*",
236
field_name: str = None,
237
) -> Callable:
238
"""
239
Register GraphQL field resolver.
240
241
Parameters:
242
- type_name: GraphQL type name or "*" for any type
243
- field_name: GraphQL field name or None for any field
244
245
Returns:
246
Decorator function for field resolver
247
"""
248
249
def batch_resolver(
250
self,
251
type_name: str = "*",
252
field_name: str = None,
253
) -> Callable:
254
"""
255
Register batch GraphQL field resolver for efficient N+1 resolution.
256
257
Parameters:
258
- type_name: GraphQL type name
259
- field_name: GraphQL field name
260
261
Returns:
262
Decorator function for batch resolver
263
"""
264
265
def resolve(self, event: Dict, context: LambdaContext) -> Any:
266
"""
267
Resolve AppSync direct Lambda resolver event.
268
269
Parameters:
270
- event: AppSync resolver event
271
- context: Lambda runtime context
272
273
Returns:
274
Resolver result (any JSON-serializable value)
275
"""
276
277
@property
278
def current_event(self) -> AppSyncResolverEvent:
279
"""Get current AppSync resolver event"""
280
281
def append_context(self, **kwargs) -> None:
282
"""Append context for resolver execution"""
283
```
284
285
### AppSync Events Resolver
286
287
Event-driven resolver for AppSync subscriptions and real-time updates.
288
289
```python { .api }
290
class AppSyncEventsResolver:
291
def __init__(self, debug: bool = None):
292
"""
293
Initialize AppSync Events resolver for subscriptions.
294
295
Parameters:
296
- debug: Enable debug mode
297
"""
298
299
def resolver(
300
self,
301
type_name: str = "*",
302
field_name: str = None,
303
) -> Callable:
304
"""Register AppSync Events field resolver"""
305
306
def resolve(self, event: Dict, context: LambdaContext) -> Any:
307
"""Resolve AppSync Events resolver event"""
308
309
@property
310
def current_event(self) -> AppSyncResolverEventsEvent:
311
"""Get current AppSync Events resolver event"""
312
```
313
314
### Lambda Function URL Resolver
315
316
HTTP request resolver for Lambda Function URLs with simplified routing.
317
318
```python { .api }
319
class LambdaFunctionUrlResolver:
320
def __init__(
321
self,
322
cors: CORSConfig = None,
323
debug: bool = None,
324
serializer: Callable[[Dict], str] = None,
325
strip_prefixes: List[str] = None,
326
enable_validation: bool = False,
327
):
328
"""
329
Initialize Lambda Function URL resolver.
330
331
Parameters:
332
- cors: CORS configuration
333
- debug: Enable debug mode
334
- serializer: Custom JSON serializer
335
- strip_prefixes: Path prefixes to strip
336
- enable_validation: Enable request validation
337
"""
338
339
# Same routing interface
340
def get(self, rule: str, **kwargs) -> Callable: ...
341
def post(self, rule: str, **kwargs) -> Callable: ...
342
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
343
344
@property
345
def current_event(self) -> LambdaFunctionUrlEvent:
346
"""Get current Function URL request event"""
347
```
348
349
### Bedrock Agent Resolver
350
351
Resolver for Amazon Bedrock Agent Lambda functions with action group integration.
352
353
```python { .api }
354
class BedrockAgentResolver:
355
def __init__(self, debug: bool = None):
356
"""
357
Initialize Bedrock Agent resolver.
358
359
Parameters:
360
- debug: Enable debug mode
361
"""
362
363
def action(self, name: str) -> Callable:
364
"""
365
Register Bedrock Agent action handler.
366
367
Parameters:
368
- name: Action name as defined in agent configuration
369
370
Returns:
371
Decorator function for action handler
372
"""
373
374
def resolve(self, event: Dict, context: LambdaContext) -> BedrockResponse:
375
"""
376
Resolve Bedrock Agent event to action handler.
377
378
Parameters:
379
- event: Bedrock Agent invocation event
380
- context: Lambda runtime context
381
382
Returns:
383
BedrockResponse object
384
"""
385
386
@property
387
def current_event(self) -> BedrockAgentEvent:
388
"""Get current Bedrock Agent event"""
389
390
class BedrockAgentFunctionResolver:
391
def __init__(self, debug: bool = None):
392
"""Initialize Bedrock Agent Function resolver"""
393
394
def function(self, name: str) -> Callable:
395
"""Register Bedrock Agent function handler"""
396
397
def resolve(self, event: Dict, context: LambdaContext) -> BedrockFunctionResponse:
398
"""Resolve Bedrock Agent function event"""
399
400
@property
401
def current_event(self) -> BedrockAgentFunctionEvent:
402
"""Get current Bedrock Agent function event"""
403
```
404
405
### VPC Lattice Resolver
406
407
Resolver for Amazon VPC Lattice service-to-service communication.
408
409
```python { .api }
410
class VPCLatticeResolver:
411
def __init__(
412
self,
413
cors: CORSConfig = None,
414
debug: bool = None,
415
serializer: Callable[[Dict], str] = None,
416
strip_prefixes: List[str] = None,
417
enable_validation: bool = False,
418
):
419
"""
420
Initialize VPC Lattice resolver.
421
422
Parameters:
423
- cors: CORS configuration
424
- debug: Enable debug mode
425
- serializer: Custom JSON serializer
426
- strip_prefixes: Path prefixes to strip
427
- enable_validation: Enable request validation
428
"""
429
430
# Same routing interface
431
def get(self, rule: str, **kwargs) -> Callable: ...
432
def post(self, rule: str, **kwargs) -> Callable: ...
433
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
434
435
@property
436
def current_event(self) -> VPCLatticeEvent:
437
"""Get current VPC Lattice event"""
438
439
class VPCLatticeV2Resolver:
440
def __init__(
441
self,
442
cors: CORSConfig = None,
443
debug: bool = None,
444
serializer: Callable[[Dict], str] = None,
445
strip_prefixes: List[str] = None,
446
enable_validation: bool = False,
447
):
448
"""Initialize VPC Lattice V2 resolver"""
449
450
def resolve(self, event: Dict, context: LambdaContext) -> Dict: ...
451
452
@property
453
def current_event(self) -> VPCLatticeEventV2:
454
"""Get current VPC Lattice V2 event"""
455
```
456
457
### Response Classes
458
459
HTTP response objects for returning structured responses from event handlers.
460
461
```python { .api }
462
class Response:
463
def __init__(
464
self,
465
status_code: int,
466
content_type: str = None,
467
body: str = None,
468
headers: Dict[str, str] = None,
469
cookies: List[str] = None,
470
compress: bool = None,
471
):
472
"""
473
Create HTTP response.
474
475
Parameters:
476
- status_code: HTTP status code
477
- content_type: Response content type
478
- body: Response body content
479
- headers: Additional HTTP headers
480
- cookies: Set-Cookie header values
481
- compress: Whether to compress response body
482
"""
483
484
class BedrockResponse:
485
def __init__(
486
self,
487
response: Dict[str, Any],
488
session_attributes: Dict[str, str] = None,
489
prompt_session_attributes: Dict[str, str] = None,
490
):
491
"""
492
Bedrock Agent response.
493
494
Parameters:
495
- response: Response data for agent
496
- session_attributes: Session attributes to persist
497
- prompt_session_attributes: Prompt session attributes
498
"""
499
500
class BedrockFunctionResponse:
501
def __init__(
502
self,
503
response: Dict[str, Any],
504
message_version: str = "1.0",
505
):
506
"""
507
Bedrock Agent function response.
508
509
Parameters:
510
- response: Function execution response
511
- message_version: Response message version
512
"""
513
```
514
515
### CORS Configuration
516
517
Cross-Origin Resource Sharing (CORS) configuration for HTTP APIs.
518
519
```python { .api }
520
class CORSConfig:
521
def __init__(
522
self,
523
allow_origin: str = "*",
524
allow_headers: List[str] = None,
525
allow_methods: List[str] = None,
526
expose_headers: List[str] = None,
527
max_age: int = None,
528
allow_credentials: bool = False,
529
):
530
"""
531
Configure CORS settings.
532
533
Parameters:
534
- allow_origin: Allowed origins for cross-origin requests
535
- allow_headers: Allowed headers in preflight requests
536
- allow_methods: Allowed HTTP methods
537
- expose_headers: Headers exposed to client
538
- max_age: Cache duration for preflight requests (seconds)
539
- allow_credentials: Whether to allow credentials
540
"""
541
```
542
543
### Middleware Support
544
545
Base classes for creating custom middleware for request/response processing.
546
547
```python { .api }
548
class BaseMiddlewareHandler:
549
def __init__(self):
550
"""Base middleware handler for event processing"""
551
552
def handler(
553
self,
554
event: Dict[str, Any],
555
context: LambdaContext,
556
next_middleware: "NextMiddleware",
557
) -> Dict[str, Any]:
558
"""
559
Process middleware logic.
560
561
Parameters:
562
- event: Incoming Lambda event
563
- context: Lambda runtime context
564
- next_middleware: Next middleware in chain
565
566
Returns:
567
Processed event or response
568
"""
569
570
class NextMiddleware:
571
def __init__(self, handler: Callable):
572
"""Next middleware wrapper"""
573
574
def __call__(
575
self,
576
event: Dict[str, Any],
577
context: LambdaContext,
578
) -> Dict[str, Any]:
579
"""Call next middleware or handler"""
580
```
581
582
## Usage Examples
583
584
### REST API with Multiple Routes
585
586
```python
587
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
588
from aws_lambda_powertools.utilities.typing import LambdaContext
589
590
app = APIGatewayRestResolver()
591
592
@app.get("/users")
593
def list_users():
594
# Access request context
595
query_params = app.current_event.query_string_parameters or {}
596
597
# Return JSON response
598
return {
599
"users": [
600
{"id": 1, "name": "Alice"},
601
{"id": 2, "name": "Bob"}
602
],
603
"total": 2,
604
"limit": query_params.get("limit", 10)
605
}
606
607
@app.post("/users")
608
def create_user():
609
# Parse JSON body
610
user_data = app.current_event.json_body
611
612
# Validate required fields
613
if not user_data.get("name"):
614
return Response(
615
status_code=400,
616
body={"error": "Name is required"}
617
)
618
619
# Create user logic here
620
new_user = {"id": 3, "name": user_data["name"]}
621
622
return Response(
623
status_code=201,
624
body=new_user
625
)
626
627
@app.get("/users/<user_id>")
628
def get_user(user_id: str):
629
# Path parameters are automatically parsed
630
# Simulate user lookup
631
if user_id == "1":
632
return {"id": 1, "name": "Alice"}
633
else:
634
return Response(status_code=404, body={"error": "User not found"})
635
636
@app.exception_handler(ValueError)
637
def handle_validation_error(ex: ValueError):
638
return Response(
639
status_code=400,
640
body={"error": f"Validation failed: {str(ex)}"}
641
)
642
643
def lambda_handler(event: dict, context: LambdaContext) -> dict:
644
return app.resolve(event, context)
645
```
646
647
### GraphQL AppSync Resolver
648
649
```python
650
from aws_lambda_powertools.event_handler import AppSyncResolver
651
from aws_lambda_powertools.utilities.typing import LambdaContext
652
653
app = AppSyncResolver()
654
655
@app.resolver(type_name="Query", field_name="getUser")
656
def get_user(user_id: str):
657
# Access GraphQL context
658
event = app.current_event
659
660
# Get arguments
661
user_id = event.arguments.get("id")
662
663
# Access identity info
664
user_identity = event.identity
665
666
# Return user data
667
return {
668
"id": user_id,
669
"name": f"User {user_id}",
670
"email": f"user{user_id}@example.com"
671
}
672
673
@app.resolver(type_name="Mutation", field_name="createUser")
674
def create_user():
675
event = app.current_event
676
user_input = event.arguments.get("input", {})
677
678
# Create user logic
679
new_user = {
680
"id": "new_123",
681
"name": user_input.get("name"),
682
"email": user_input.get("email")
683
}
684
685
return new_user
686
687
@app.batch_resolver(type_name="User", field_name="posts")
688
def get_user_posts(user_ids: List[str]):
689
# Batch resolver for efficient N+1 resolution
690
# This runs once for multiple User.posts requests
691
692
# Bulk fetch posts for all users
693
posts_by_user = fetch_posts_for_users(user_ids)
694
695
# Return posts in same order as user_ids
696
return [posts_by_user.get(user_id, []) for user_id in user_ids]
697
698
def lambda_handler(event: dict, context: LambdaContext):
699
return app.resolve(event, context)
700
```
701
702
### Bedrock Agent with Actions
703
704
```python
705
from aws_lambda_powertools.event_handler import BedrockAgentResolver, BedrockResponse
706
from aws_lambda_powertools.utilities.typing import LambdaContext
707
708
app = BedrockAgentResolver()
709
710
@app.action("get_weather")
711
def get_weather():
712
event = app.current_event
713
714
# Extract parameters from agent
715
parameters = event.parameters
716
location = parameters.get("location", "unknown")
717
718
# Get weather data (mock)
719
weather_data = {
720
"location": location,
721
"temperature": "72°F",
722
"conditions": "sunny",
723
"forecast": "Clear skies expected"
724
}
725
726
return BedrockResponse(
727
response=weather_data,
728
session_attributes={"last_location": location}
729
)
730
731
@app.action("book_appointment")
732
def book_appointment():
733
event = app.current_event
734
735
parameters = event.parameters
736
date = parameters.get("date")
737
time = parameters.get("time")
738
service = parameters.get("service")
739
740
# Appointment booking logic
741
booking_result = {
742
"confirmation_id": "BOOK123",
743
"date": date,
744
"time": time,
745
"service": service,
746
"status": "confirmed"
747
}
748
749
return BedrockResponse(
750
response=booking_result,
751
session_attributes={
752
"last_booking": "BOOK123",
753
"user_preferences": {"service": service}
754
}
755
)
756
757
def lambda_handler(event: dict, context: LambdaContext):
758
return app.resolve(event, context)
759
```
760
761
### CORS and Error Handling
762
763
```python
764
from aws_lambda_powertools.event_handler import (
765
APIGatewayRestResolver,
766
CORSConfig,
767
Response
768
)
769
from aws_lambda_powertools.utilities.typing import LambdaContext
770
771
# Configure CORS
772
cors_config = CORSConfig(
773
allow_origin="https://myapp.example.com",
774
allow_headers=["Content-Type", "Authorization"],
775
allow_methods=["GET", "POST", "PUT", "DELETE"],
776
expose_headers=["X-Custom-Header"],
777
max_age=3600,
778
allow_credentials=True
779
)
780
781
app = APIGatewayRestResolver(cors=cors_config)
782
783
@app.get("/api/data")
784
def get_data():
785
return {"message": "Hello from API"}
786
787
@app.exception_handler(ValueError)
788
def handle_validation_error(ex: ValueError):
789
return Response(
790
status_code=400,
791
body={"error": "Invalid input", "details": str(ex)},
792
headers={"X-Error-Type": "ValidationError"}
793
)
794
795
@app.exception_handler(KeyError)
796
def handle_missing_key(ex: KeyError):
797
return Response(
798
status_code=400,
799
body={"error": "Missing required field", "field": str(ex)}
800
)
801
802
@app.not_found()
803
def handle_not_found():
804
return Response(
805
status_code=404,
806
body={"error": "Resource not found"}
807
)
808
809
@app.unauthorized()
810
def handle_unauthorized():
811
return Response(
812
status_code=401,
813
body={"error": "Authentication required"},
814
headers={"WWW-Authenticate": "Bearer"}
815
)
816
817
def lambda_handler(event: dict, context: LambdaContext) -> dict:
818
return app.resolve(event, context)
819
```
820
821
## Types
822
823
```python { .api }
824
from typing import Dict, List, Any, Union, Callable, Optional
825
from aws_lambda_powertools.utilities.data_classes import (
826
APIGatewayProxyEvent,
827
APIGatewayProxyEventV2,
828
ALBEvent,
829
AppSyncResolverEvent,
830
AppSyncResolverEventsEvent,
831
BedrockAgentEvent,
832
BedrockAgentFunctionEvent,
833
LambdaFunctionUrlEvent,
834
VPCLatticeEvent,
835
VPCLatticeEventV2,
836
)
837
838
# Route handler function signature
839
RouteHandler = Callable[..., Union[Dict[str, Any], Response, Any]]
840
841
# Exception handler function signature
842
ExceptionHandler = Callable[[Exception], Union[Dict[str, Any], Response]]
843
844
# Middleware function signature
845
MiddlewareHandler = Callable[
846
[Dict[str, Any], LambdaContext, NextMiddleware],
847
Dict[str, Any]
848
]
849
850
# Serializer function signature
851
Serializer = Callable[[Dict[str, Any]], str]
852
```