or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

Files

docs

batch-processing.mdcore-observability.mddata-classes.mdevent-handlers.mdfeature-flags.mdindex.mdparameters.mdparser.mdutilities.md

event-handlers.mddocs/

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

```