0
# Plugins and Extensions
1

2
LocalStack provides an extensible plugin system that enables custom CLI commands, AWS service providers, runtime extensions, and package management. The plugin architecture allows for modular functionality while maintaining compatibility with LocalStack's core systems.
3

4
## Capabilities
5

6
### CLI Plugin System
7

8
Framework for extending the LocalStack CLI with custom commands and functionality.
9

10
```python { .api }
11
class LocalstackCliPlugin:
12
    """
13
    Base class for LocalStack CLI plugins.
14
    
15
    CLI plugins can add custom commands, modify existing commands,
16
    and integrate with LocalStack's configuration system.
17
    
18
    Location: localstack.cli.plugin
19
    """
20
    
21
    namespace: str = "localstack.plugins.cli"
22
    
23
    def attach(self, cli) -> None:
24
        """
25
        Attach plugin commands and functionality to CLI.
26
        
27
        Called during CLI initialization to register custom commands,
28
        modify existing commands, or add global options.
29
        
30
        Args:
31
            cli: LocalStack CLI instance for command registration
32
        """
33

34
def load_cli_plugins(cli) -> None:
35
    """
36
    Discover and load all CLI plugins from the namespace.
37
    
38
    Automatically called during CLI initialization to load
39
    all plugins registered in the 'localstack.plugins.cli' namespace.
40
    
41
    Args:
42
        cli: LocalStack CLI instance
43
        
44
    Location: localstack.cli.plugin
45
    """
46
```
47

48
### Service Plugin System
49

50
Plugin framework for implementing custom AWS service providers and extending existing services.
51

52
```python { .api }
53
class ServicePlugin:
54
    """
55
    Plugin wrapper for AWS service provider implementations.
56
    
57
    Service plugins enable custom AWS service implementations
58
    that integrate with LocalStack's service management system.
59
    
60
    Location: localstack.services.plugins
61
    """
62
    
63
    service: str        # AWS service name (e.g., 'myservice')
64
    api: str           # AWS API identifier
65
    
66
    def create_service(self) -> 'Service':
67
        """
68
        Create service instance from this plugin.
69
        
70
        Returns:
71
            Configured Service instance with provider implementation
72
        """
73

74
def aws_provider(
75
    api: str = None, 
76
    name: str = "default", 
77
    should_load: callable = None
78
) -> callable:
79
    """
80
    Decorator for registering AWS service provider implementations.
81
    
82
    Use this decorator to mark classes as AWS service providers
83
    that implement specific AWS APIs.
84
    
85
    Args:
86
        api: AWS service API name (e.g., 's3', 'lambda', 'myservice')
87
        name: Provider name for multiple implementations (default: "default")
88
        should_load: Optional function returning bool to control loading
89
        
90
    Returns:
91
        Decorator function for provider classes
92
        
93
    Location: localstack.services.plugins
94
    """
95

96
class ServiceProvider(Protocol):
97
    """
98
    Protocol interface for AWS service provider implementations.
99
    
100
    Service providers implement AWS service operations and integrate
101
    with LocalStack's AWS Service Framework (ASF).
102
    
103
    Location: localstack.services.plugins
104
    """
105
    
106
    service: str  # AWS service name this provider implements
107

108
# Plugin namespace for AWS service providers
109
PLUGIN_NAMESPACE: str = "localstack.aws.provider"
110
```
111

112
### Extension System
113

114
Runtime extension loading and management for enhanced LocalStack functionality.
115

116
```python { .api }
117
class Extension:
118
    """
119
    Base class for LocalStack runtime extensions.
120
    
121
    Extensions can modify LocalStack behavior, add new capabilities,
122
    and integrate with the service lifecycle.
123
    
124
    Location: localstack.extensions
125
    """
126
    
127
    name: str           # Extension name
128
    version: str        # Extension version
129
    
130
    def on_extension_load(self) -> None:
131
        """Called when extension is loaded during startup"""
132
        
133
    def on_platform_start(self) -> None:
134
        """Called when LocalStack platform starts"""
135
        
136
    def on_platform_shutdown(self) -> None:
137
        """Called when LocalStack platform shuts down"""
138

139
def load_extensions() -> list['Extension']:
140
    """
141
    Discover and load all registered extensions.
142
    
143
    Returns:
144
        List of loaded extension instances
145
        
146
    Location: localstack.extensions
147
    """
148

149
def register_extension(extension_class: type) -> None:
150
    """
151
    Register extension class for automatic loading.
152
    
153
    Args:
154
        extension_class: Extension class to register
155
        
156
    Location: localstack.extensions
157
    """
158
```
159

160
### Package Management System
161

162
Package plugin management for third-party integrations and additional services.
163

164
```python { .api }
165
class PackagePlugin:
166
    """
167
    Plugin for managing third-party packages and integrations.
168
    
169
    Package plugins enable installation and management of
170
    external tools, services, and dependencies.
171
    
172
    Location: localstack.packages
173
    """
174
    
175
    name: str               # Package name
176
    version: str            # Package version
177
    dependencies: list[str] # Package dependencies
178
    
179
    def install(self, target_dir: str) -> bool:
180
        """
181
        Install package to target directory.
182
        
183
        Args:
184
            target_dir: Installation target directory
185
            
186
        Returns:
187
            True if installation successful
188
        """
189
        
190
    def is_installed(self) -> bool:
191
        """
192
        Check if package is already installed.
193
        
194
        Returns:
195
            True if package is available
196
        """
197
        
198
    def get_version(self) -> str:
199
        """
200
        Get installed package version.
201
        
202
        Returns:
203
            Version string or None if not installed
204
        """
205

206
def install_package(
207
    package_name: str, 
208
    version: str = None,
209
    target: str = None
210
) -> bool:
211
    """
212
    Install package using package manager.
213
    
214
    Args:
215
        package_name: Name of package to install
216
        version: Specific version to install (None for latest)
217
        target: Installation target directory
218
        
219
    Returns:
220
        True if installation successful
221
        
222
    Location: localstack.packages
223
    """
224

225
def list_available_packages() -> list[dict[str, str]]:
226
    """
227
    List packages available for installation.
228
    
229
    Returns:
230
        List of package information dictionaries
231
        
232
    Location: localstack.packages
233
    """
234
```
235

236
## Plugin Development
237

238
### Creating CLI Plugins
239

240
Example implementation of a custom CLI plugin:
241

242
```python
243
from localstack.cli.plugin import LocalstackCliPlugin
244
import click
245

246
class MyCliPlugin(LocalstackCliPlugin):
247
    """Custom CLI plugin example"""
248
    
249
    def attach(self, cli):
250
        """Add custom commands to CLI"""
251
        
252
        @cli.group()
253
        def mycommands():
254
            """Custom command group"""
255
            pass
256
            
257
        @mycommands.command()
258
        @click.option('--target', help='Target for operation')
259
        def deploy(target):
260
            """Deploy custom resources"""
261
            click.echo(f"Deploying to {target}")
262
            # Custom deployment logic
263
            
264
        @mycommands.command() 
265
        def status():
266
            """Check custom resource status"""
267
            click.echo("Checking status...")
268
            # Custom status logic
269

270
# Register plugin via entry point in setup.py/pyproject.toml:
271
# [project.entry-points."localstack.plugins.cli"]
272
# my-plugin = "mypackage.cli:MyCliPlugin"
273
```
274

275
### Creating Service Providers
276

277
Example implementation of a custom AWS service provider:
278

279
```python
280
from localstack.services.plugins import aws_provider, ServiceProvider
281
from localstack.aws.api import RequestContext
282
from typing import Dict, Any
283

284
@aws_provider(api="myservice")
285
class MyServiceProvider(ServiceProvider):
286
    """Custom AWS service provider"""
287
    
288
    service = "myservice"
289
    
290
    def create_resource(
291
        self, 
292
        context: RequestContext, 
293
        request: Dict[str, Any]
294
    ) -> Dict[str, Any]:
295
        """
296
        Create a custom resource.
297
        
298
        Args:
299
            context: Request context with AWS metadata
300
            request: Parsed request parameters
301
            
302
        Returns:
303
            AWS-compatible response dictionary
304
        """
305
        resource_name = request.get("ResourceName")
306
        resource_id = f"resource-{self._generate_id()}"
307
        
308
        # Store resource (implementation specific)
309
        self._store_resource(resource_id, {
310
            "ResourceId": resource_id,
311
            "ResourceName": resource_name,
312
            "Status": "ACTIVE"
313
        })
314
        
315
        return {
316
            "ResourceId": resource_id,
317
            "ResourceArn": f"arn:aws:myservice:us-east-1:000000000000:resource/{resource_id}"
318
        }
319
    
320
    def list_resources(
321
        self, 
322
        context: RequestContext, 
323
        request: Dict[str, Any]
324
    ) -> Dict[str, Any]:
325
        """List all resources"""
326
        resources = self._list_stored_resources()
327
        
328
        return {
329
            "Resources": [
330
                {
331
                    "ResourceId": r["ResourceId"],
332
                    "ResourceName": r["ResourceName"], 
333
                    "Status": r["Status"]
334
                }
335
                for r in resources
336
            ]
337
        }
338
    
339
    def _generate_id(self) -> str:
340
        """Generate unique resource ID"""
341
        import uuid
342
        return str(uuid.uuid4())[:8]
343
        
344
    def _store_resource(self, resource_id: str, resource: Dict[str, Any]) -> None:
345
        """Store resource (implement with your preferred storage)"""
346
        pass
347
        
348
    def _list_stored_resources(self) -> List[Dict[str, Any]]:
349
        """List stored resources"""
350
        return []
351

352
# Register via entry point:
353
# [project.entry-points."localstack.aws.provider"]  
354
# myservice = "mypackage.services:MyServiceProvider"
355
```
356

357
### Creating Extensions
358

359
Example runtime extension that adds custom functionality:
360

361
```python
362
from localstack.extensions import Extension
363
import logging
364

365
LOG = logging.getLogger(__name__)
366

367
class MyExtension(Extension):
368
    """Custom LocalStack extension"""
369
    
370
    name = "my-extension"
371
    version = "1.0.0"
372
    
373
    def on_extension_load(self):
374
        """Initialize extension on load"""
375
        LOG.info("Loading my custom extension")
376
        self._setup_custom_functionality()
377
        
378
    def on_platform_start(self):
379
        """Execute when LocalStack starts"""
380
        LOG.info("LocalStack platform starting - extension active")
381
        self._start_background_tasks()
382
        
383
    def on_platform_shutdown(self):
384
        """Cleanup when LocalStack stops"""
385
        LOG.info("LocalStack platform shutting down")
386
        self._cleanup_resources()
387
        
388
    def _setup_custom_functionality(self):
389
        """Setup extension-specific functionality"""
390
        # Add custom middleware, modify configs, etc.
391
        pass
392
        
393
    def _start_background_tasks(self):
394
        """Start extension background tasks"""
395
        # Start monitoring, periodic tasks, etc.
396
        pass
397
        
398
    def _cleanup_resources(self):
399
        """Clean up extension resources"""
400
        # Stop tasks, close connections, etc.
401
        pass
402

403
# Register extension
404
from localstack.extensions import register_extension
405
register_extension(MyExtension)
406
```
407

408
### Creating Package Plugins
409

410
Example package plugin for managing external dependencies:
411

412
```python
413
from localstack.packages import PackagePlugin
414
import subprocess
415
import os
416

417
class RedisPackage(PackagePlugin):
418
    """Redis server package plugin"""
419
    
420
    name = "redis"
421
    version = "7.0.0"
422
    dependencies = []
423
    
424
    def install(self, target_dir: str) -> bool:
425
        """Install Redis server"""
426
        try:
427
            # Download and install Redis
428
            redis_url = f"https://download.redis.io/releases/redis-{self.version}.tar.gz"
429
            
430
            # Create target directory
431
            os.makedirs(target_dir, exist_ok=True)
432
            
433
            # Download, extract, and build Redis
434
            subprocess.run([
435
                "curl", "-L", redis_url, "|", 
436
                "tar", "xz", "-C", target_dir
437
            ], check=True)
438
            
439
            redis_dir = os.path.join(target_dir, f"redis-{self.version}")
440
            subprocess.run(["make"], cwd=redis_dir, check=True)
441
            
442
            return True
443
        except Exception as e:
444
            print(f"Redis installation failed: {e}")
445
            return False
446
            
447
    def is_installed(self) -> bool:
448
        """Check if Redis is installed"""
449
        try:
450
            subprocess.run(["redis-server", "--version"], 
451
                         check=True, capture_output=True)
452
            return True
453
        except (subprocess.CalledProcessError, FileNotFoundError):
454
            return False
455
            
456
    def get_version(self) -> str:
457
        """Get installed Redis version"""
458
        try:
459
            result = subprocess.run(
460
                ["redis-server", "--version"],
461
                capture_output=True, text=True, check=True
462
            )
463
            # Parse version from output
464
            return result.stdout.split()[2]
465
        except Exception:
466
            return None
467

468
# Register via entry point:
469
# [project.entry-points."localstack.packages"]
470
# redis = "mypackage.packages:RedisPackage"
471
```
472

473
## Plugin Discovery and Loading
474

475
### Entry Point Configuration
476

477
Plugins are discovered via Python entry points in `pyproject.toml`:
478

479
```toml
480
[project.entry-points."localstack.plugins.cli"]
481
my-cli-plugin = "mypackage.cli:MyCliPlugin"
482

483
[project.entry-points."localstack.aws.provider"] 
484
myservice = "mypackage.services:MyServiceProvider"
485
custom-s3 = "mypackage.s3:CustomS3Provider"
486

487
[project.entry-points."localstack.extensions"]
488
my-extension = "mypackage.extensions:MyExtension"
489

490
[project.entry-points."localstack.packages"]
491
redis = "mypackage.packages:RedisPackage"
492
elasticsearch = "mypackage.packages:ElasticsearchPackage"
493
```
494

495
### Plugin Loading Process
496

497
```python { .api }
498
def discover_plugins(namespace: str) -> list[object]:
499
    """
500
    Discover plugins from entry point namespace.
501
    
502
    Args:
503
        namespace: Entry point namespace to search
504
        
505
    Returns:
506
        List of plugin instances
507
    """
508

509
def load_plugin(plugin_class: type, config: dict = None) -> object:
510
    """
511
    Load and initialize plugin instance.
512
    
513
    Args:
514
        plugin_class: Plugin class to instantiate
515
        config: Optional plugin configuration
516
        
517
    Returns:
518
        Initialized plugin instance
519
    """
520
```
521

522
### Plugin Configuration
523

524
Plugins can access LocalStack configuration and define their own settings:
525

526
```python
527
class ConfigurablePlugin(LocalstackCliPlugin):
528
    """Plugin with configuration support"""
529
    
530
    def __init__(self):
531
        self.config = self._load_config()
532
    
533
    def _load_config(self) -> dict:
534
        """Load plugin-specific configuration"""
535
        from localstack.config import config
536
        
537
        return {
538
            "enabled": config.get("MY_PLUGIN_ENABLED", "true").lower() == "true",
539
            "api_key": config.get("MY_PLUGIN_API_KEY"),
540
            "endpoint": config.get("MY_PLUGIN_ENDPOINT", "https://api.example.com")
541
        }
542
    
543
    def attach(self, cli):
544
        if not self.config["enabled"]:
545
            return  # Skip if disabled
546
            
547
        @cli.command()
548
        def my_command():
549
            """Custom command with configuration"""
550
            api_key = self.config["api_key"]
551
            endpoint = self.config["endpoint"]
552
            # Use configuration in command logic
553
```
554

555
## Plugin Distribution
556

557
### Package Structure
558

559
Recommended structure for plugin packages:
560

561
```
562
my-localstack-plugin/
563
├── pyproject.toml              # Package configuration with entry points
564
├── README.md                   # Plugin documentation
565
├── my_plugin/
566
│   ├── __init__.py
567
│   ├── cli.py                  # CLI plugin implementation
568
│   ├── services.py             # Service provider implementations  
569
│   ├── extensions.py           # Runtime extensions
570
│   └── packages.py             # Package managers
571
└── tests/
572
    ├── test_cli.py
573
    ├── test_services.py
574
    └── test_extensions.py
575
```
576

577
### Installation and Usage
578

579
Users install plugins as regular Python packages:
580

581
```bash
582
# Install plugin
583
pip install my-localstack-plugin
584

585
# Plugin commands are automatically available
586
localstack my-custom-command --help
587

588
# Plugin services are automatically loaded
589
localstack start  # Custom services included
590

591
# Plugin packages available via LPM
592
localstack lpm install my-custom-package
593
```
594

595
### Plugin Testing
596

597
Test plugins using LocalStack's testing framework:
598

599
```python
600
import pytest
601
from localstack.testing.pytest import localstack
602

603
def test_my_plugin_cli():
604
    """Test custom CLI command"""
605
    from click.testing import CliRunner
606
    from my_plugin.cli import MyCliPlugin
607
    
608
    runner = CliRunner()
609
    # Test CLI plugin functionality
610

611
@pytest.mark.aws  
612
def test_my_service_provider(localstack):
613
    """Test custom service provider"""
614
    import boto3
615
    
616
    # Create client for custom service
617
    client = boto3.client(
618
        'myservice',
619
        endpoint_url='http://localhost:4566',
620
        aws_access_key_id='test',
621
        aws_secret_access_key='test'
622
    )
623
    
624
    # Test service operations
625
    response = client.create_resource(ResourceName='test')
626
    assert 'ResourceId' in response
627
```