MiniStack AWS Emulator
Skill by ara.so — Daily 2026 Skills collection.
MiniStack is a free, MIT-licensed drop-in replacement for LocalStack that emulates 25+ AWS services (S3, SQS, DynamoDB, Lambda, SNS, IAM, STS, Kinesis, EventBridge, SecretsManager, SSM, CloudWatch, SES, and more) on a single port (4566). No account, no API key, no telemetry. Works with boto3, AWS CLI, Terraform, CDK, and any SDK.
Installation
Option 1: PyPI (simplest)
bash
pip install ministackministack# Server runs at http://localhost:4566# Change port: GATEWAY_PORT=5000 ministackOption 2: Docker Hub
bash
docker run -p 4566:4566 nahuelnucera/ministackOption 3: Docker Compose (from source)
bash
git clone https://github.com/Nahuel990/ministackcd ministackdocker compose up -dVerify it's running
bash
curl http://localhost:4566/_localstack/healthConfiguration
| Environment Variable | Default | Description |
|---|---|---|
GATEWAY_PORT | 4566 | Port to listen on |
S3_PERSIST | 0 | Set to 1 to persist S3 data to disk |
AWS CLI Usage
bash
# Set credentials (any non-empty values work)export AWS_ACCESS_KEY_ID=testexport AWS_SECRET_ACCESS_KEY=testexport AWS_DEFAULT_REGION=us-east-1
# S3aws --endpoint-url=http://localhost:4566 s3 mb s3://my-bucketaws --endpoint-url=http://localhost:4566 s3 cp ./file.txt s3://my-bucket/aws --endpoint-url=http://localhost:4566 s3 ls s3://my-bucket
# SQSaws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queueaws --endpoint-url=http://localhost:4566 sqs list-queues
# DynamoDBaws --endpoint-url=http://localhost:4566 dynamodb list-tablesaws --endpoint-url=http://localhost:4566 dynamodb create-table \ --table-name Users \ --attribute-definitions AttributeName=userId,AttributeType=S \ --key-schema AttributeName=userId,KeyType=HASH \ --billing-mode PAY_PER_REQUEST
# STS (identity check)aws --endpoint-url=http://localhost:4566 sts get-caller-identity
# Use a named profile insteadaws configure --profile local# Enter: test / test / us-east-1 / jsonaws --profile local --endpoint-url=http://localhost:4566 s3 lsawslocal wrapper (from source)
bash
chmod +x bin/awslocal./bin/awslocal s3 ls./bin/awslocal dynamodb list-tablesboto3 Usage Patterns
Universal client factory
python
import boto3
ENDPOINT = "http://localhost:4566"
def aws_client(service: str): return boto3.client( service, endpoint_url=ENDPOINT, aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1", )
def aws_resource(service: str): return boto3.resource( service, endpoint_url=ENDPOINT, aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1", )S3
python
s3 = aws_client("s3")
# Create bucket and uploads3.create_bucket(Bucket="my-bucket")s3.put_object(Bucket="my-bucket", Key="hello.txt", Body=b"Hello, MiniStack!")
# Downloadobj = s3.get_object(Bucket="my-bucket", Key="hello.txt")print(obj["Body"].read()) # b'Hello, MiniStack!'
# List objectsresponse = s3.list_objects_v2(Bucket="my-bucket")for item in response.get("Contents", []): print(item["Key"])
# Copy objects3.copy_object( Bucket="my-bucket", CopySource={"Bucket": "my-bucket", "Key": "hello.txt"}, Key="hello-copy.txt",)
# Enable versionings3.put_bucket_versioning( Bucket="my-bucket", VersioningConfiguration={"Status": "Enabled"},)
# Presigned URL (works locally)url = s3.generate_presigned_url( "get_object", Params={"Bucket": "my-bucket", "Key": "hello.txt"}, ExpiresIn=3600,)SQS
python
sqs = aws_client("sqs")
# Standard queuequeue = sqs.create_queue(QueueName="my-queue")queue_url = queue["QueueUrl"]
sqs.send_message(QueueUrl=queue_url, MessageBody='{"event": "user_signup"}')
messages = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=10)for msg in messages.get("Messages", []): print(msg["Body"]) sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])
# FIFO queuefifo = sqs.create_queue( QueueName="my-queue.fifo", Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},)
# Dead-letter queue setupdlq = sqs.create_queue(QueueName="my-dlq")dlq_attrs = sqs.get_queue_attributes( QueueUrl=dlq["QueueUrl"], AttributeNames=["QueueArn"])sqs.set_queue_attributes( QueueUrl=queue_url, Attributes={ "RedrivePolicy": json.dumps({ "deadLetterTargetArn": dlq_attrs["Attributes"]["QueueArn"], "maxReceiveCount": "3", }) },)DynamoDB
python
import jsonddb = aws_client("dynamodb")
# Create tableddb.create_table( TableName="Users", KeySchema=[ {"AttributeName": "userId", "KeyType": "HASH"}, {"AttributeName": "createdAt", "KeyType": "RANGE"}, ], AttributeDefinitions=[ {"AttributeName": "userId", "AttributeType": "S"}, {"AttributeName": "createdAt", "AttributeType": "N"}, ], BillingMode="PAY_PER_REQUEST",)
# Put / Get / Deleteddb.put_item( TableName="Users", Item={ "userId": {"S": "u1"}, "createdAt": {"N": "1700000000"}, "name": {"S": "Alice"}, "active": {"BOOL": True}, },)
item = ddb.get_item( TableName="Users", Key={"userId": {"S": "u1"}, "createdAt": {"N": "1700000000"}},)print(item["Item"]["name"]["S"]) # Alice
# Queryresult = ddb.query( TableName="Users", KeyConditionExpression="userId = :uid", ExpressionAttributeValues={":uid": {"S": "u1"}},)
# Batch writeddb.batch_write_item( RequestItems={ "Users": [ {"PutRequest": {"Item": {"userId": {"S": "u2"}, "createdAt": {"N": "1700000001"}, "name": {"S": "Bob"}}}}, ] })
# TTLddb.update_time_to_live( TableName="Users", TimeToLiveSpecification={"Enabled": True, "AttributeName": "expiresAt"},)SNS + SQS Fanout
python
sns = aws_client("sns")sqs = aws_client("sqs")
topic = sns.create_topic(Name="my-topic")topic_arn = topic["TopicArn"]
queue = sqs.create_queue(QueueName="fan-queue")queue_attrs = sqs.get_queue_attributes( QueueUrl=queue["QueueUrl"], AttributeNames=["QueueArn"])queue_arn = queue_attrs["Attributes"]["QueueArn"]
sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)
# Publish — message is fanned out to subscribed SQS queuessns.publish(TopicArn=topic_arn, Message="hello fanout", Subject="test")Lambda
python
import zipfile, io
# Create a zip with handler codebuf = io.BytesIO()with zipfile.ZipFile(buf, "w") as zf: zf.writestr("handler.py", """def handler(event, context): print("event:", event) return {"statusCode": 200, "body": "ok"}""")buf.seek(0)
lam = aws_client("lambda")
lam.create_function( FunctionName="my-function", Runtime="python3.12", Role="arn:aws:iam::000000000000:role/role", Handler="handler.handler", Code={"ZipFile": buf.read()},)
# Invoke synchronouslyimport jsonresponse = lam.invoke( FunctionName="my-function", InvocationType="RequestResponse", Payload=json.dumps({"key": "value"}),)result = json.loads(response["Payload"].read())print(result) # {"statusCode": 200, "body": "ok"}
# SQS event source mappinglam.create_event_source_mapping( EventSourceArn=queue_arn, FunctionName="my-function", BatchSize=10, Enabled=True,)SecretsManager
python
sm = aws_client("secretsmanager")
sm.create_secret(Name="db-password", SecretString='{"password":"s3cr3t"}')secret = sm.get_secret_value(SecretId="db-password")print(secret["SecretString"]) # {"password":"s3cr3t"}
sm.update_secret(SecretId="db-password", SecretString='{"password":"newpass"}')sm.delete_secret(SecretId="db-password", ForceDeleteWithoutRecovery=True)SSM Parameter Store
python
ssm = aws_client("ssm")
ssm.put_parameter(Name="/app/db/host", Value="localhost", Type="String")ssm.put_parameter(Name="/app/db/password", Value="secret", Type="SecureString")
param = ssm.get_parameter(Name="/app/db/host")print(param["Parameter"]["Value"]) # localhost
# Fetch all params under a pathparams = ssm.get_parameters_by_path(Path="/app/", Recursive=True)for p in params["Parameters"]: print(p["Name"], p["Value"])Kinesis
python
import base64
kin = aws_client("kinesis")
kin.create_stream(StreamName="events", ShardCount=1)kin.put_record(StreamName="events", Data=b'{"event":"click"}', PartitionKey="user1")
# Get recordsshards = kin.list_shards(StreamName="events")shard_id = shards["Shards"][0]["ShardId"]
iterator = kin.get_shard_iterator( StreamName="events", ShardId=shard_id, ShardIteratorType="TRIM_HORIZON",)records = kin.get_records(ShardIterator=iterator["ShardIterator"])for r in records["Records"]: print(base64.b64decode(r["Data"]))EventBridge
python
eb = aws_client("events")
# Create a custom buseb.create_event_bus(Name="my-bus")
# Put a rule targeting a Lambdaeb.put_rule( Name="my-rule", EventBusName="my-bus", EventPattern='{"source": ["myapp"]}', State="ENABLED",)eb.put_targets( Rule="my-rule", EventBusName="my-bus", Targets=[{"Id": "1", "Arn": "arn:aws:lambda:us-east-1:000000000000:function:my-function"}],)
# Emit an event (triggers Lambda target)eb.put_events(Entries=[{ "Source": "myapp", "DetailType": "UserSignup", "Detail": '{"userId": "123"}', "EventBusName": "my-bus",}])CloudWatch Logs
python
import time
logs = aws_client("logs")
logs.create_log_group(logGroupName="/app/service")logs.create_log_stream(logGroupName="/app/service", logStreamName="stream-1")
logs.put_log_events( logGroupName="/app/service", logStreamName="stream-1", logEvents=[ {"timestamp": int(time.time() * 1000), "message": "App started"}, {"timestamp": int(time.time() * 1000), "message": "Request received"}, ],)
events = logs.get_log_events( logGroupName="/app/service", logStreamName="stream-1",)for e in events["events"]: print(e["message"])
# Filter with glob patterns (* and ?), AND terms, -exclusionsfiltered = logs.filter_log_events( logGroupName="/app/service", filterPattern="Request*",)Testing Patterns
pytest fixture (recommended)
python
import pytestimport boto3
MINISTACK_ENDPOINT = "http://localhost:4566"
@pytest.fixture(scope="session")def aws_endpoint(): return MINISTACK_ENDPOINT
@pytest.fixturedef s3_client(aws_endpoint): return boto3.client( "s3", endpoint_url=aws_endpoint, aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1", )
@pytest.fixturedef test_bucket(s3_client): bucket = "test-bucket" s3_client.create_bucket(Bucket=bucket) yield bucket # Cleanup objs = s3_client.list_objects_v2(Bucket=bucket).get("Contents", []) for obj in objs: s3_client.delete_object(Bucket=bucket, Key=obj["Key"]) s3_client.delete_bucket(Bucket=bucket)
def test_upload_download(s3_client, test_bucket): s3_client.put_object(Bucket=test_bucket, Key="test.txt", Body=b"hello") resp = s3_client.get_object(Bucket=test_bucket, Key="test.txt") assert resp["Body"].read() == b"hello"GitHub Actions CI integration
yaml
# .github/workflows/test.ymlname: Test
on: [push, pull_request]
jobs: test: runs-on: ubuntu-latest services: ministack: image: nahuelnucera/ministack ports: - 4566:4566 steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.12" - run: pip install -r requirements.txt - run: pytest env: AWS_ACCESS_KEY_ID: test AWS_SECRET_ACCESS_KEY: test AWS_DEFAULT_REGION: us-east-1 AWS_ENDPOINT_URL: http://localhost:4566Using AWS_ENDPOINT_URL env var (boto3 >= 1.28)
python
import osimport boto3
# If AWS_ENDPOINT_URL is set, boto3 uses it automatically — no endpoint_url kwarg needed# export AWS_ENDPOINT_URL=http://localhost:4566s3 = boto3.client("s3") # picks up AWS_ENDPOINT_URL automaticallySupported Services (25+)
| Service | Key Operations |
|---|---|
| S3 | CRUD, multipart, versioning, encryption, lifecycle, CORS, ACL, notifications |
| SQS | Standard & FIFO queues, DLQ, batch ops |
| SNS | Topics, subscriptions, fanout to SQS/Lambda, platform endpoints |
| DynamoDB | Tables, CRUD, Query, Scan, TTL, transactions, batch ops |
| Lambda | Python runtimes, invoke, SQS event sources, Function URLs |
| IAM | Users, roles, policies, groups, instance profiles, OIDC |
| STS | GetCallerIdentity, AssumeRole, GetSessionToken |
| SecretsManager | Full CRUD, rotation, versioning |
| SSM Parameter Store | String, SecureString, StringList, path queries |
| EventBridge | Buses, rules, targets, Lambda dispatch |
| Kinesis | Streams, shards, records, iterators |
| CloudWatch Metrics | PutMetricData, alarms, dashboards, CBOR protocol |
| CloudWatch Logs | Log groups/streams, filter with globs, metric filters |
| SES | Send email, templates, configuration sets |
| Step Functions | State machine CRUD |
| RDS | Spins up real Postgres/MySQL containers |
| ElastiCache | Spins up real Redis containers |
| Athena | Real SQL via DuckDB |
| ECS | Real Docker containers |
Troubleshooting
Connection refused on port 4566
bash
# Check if ministack is runningcurl http://localhost:4566/_localstack/health# Start itministack# ordocker run -p 4566:4566 nahuelnucera/ministackNoCredentialsError from boto3
bash
export AWS_ACCESS_KEY_ID=testexport AWS_SECRET_ACCESS_KEY=testexport AWS_DEFAULT_REGION=us-east-1# Any non-empty values work — MiniStack doesn't validate credentialsInvalidSignatureException
- This is usually a region mismatch. Ensure
region_name="us-east-1"matches across all clients.
Lambda function not found after create
- MiniStack executes Python runtimes with a warm worker pool. Wait briefly or invoke with
InvocationType="Event"for async.
S3 data lost on restart
bash
# Enable persistenceS3_PERSIST=1 ministack# or in Dockerdocker run -p 4566:4566 -e S3_PERSIST=1 -v $(pwd)/data:/data nahuelnucera/ministackPort conflict
bash
GATEWAY_PORT=5000 ministack# Then use http://localhost:5000 as endpointMigrating from LocalStack
- Replace all
http://localhost:4566endpoint URLs — they stay the same. - Remove
LOCALSTACK_AUTH_TOKEN/LOCALSTACK_API_KEYenv vars (not needed). - Replace
localstack/localstackDocker image withnahuelnucera/ministack. - All
boto3client code works without modification.
