Ministack Aws Emulator

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)

pip install ministack
ministack
# Server runs at http://localhost:4566
# Change port: GATEWAY_PORT=5000 ministack

Option 2: Docker Hub

docker run -p 4566:4566 nahuelnucera/ministack

Option 3: Docker Compose (from source)

git clone https://github.com/Nahuel990/ministack
cd ministack
docker compose up -d

Verify it's running

curl http://localhost:4566/_localstack/health

---

Configuration

---

AWS CLI Usage

# Set credentials (any non-empty values work)
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1

# S3
aws --endpoint-url=http://localhost:4566 s3 mb s3://my-bucket
aws --endpoint-url=http://localhost:4566 s3 cp ./file.txt s3://my-bucket/
aws --endpoint-url=http://localhost:4566 s3 ls s3://my-bucket

# SQS
aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queue
aws --endpoint-url=http://localhost:4566 sqs list-queues

# DynamoDB
aws --endpoint-url=http://localhost:4566 dynamodb list-tables
aws --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 instead
aws configure --profile local
# Enter: test / test / us-east-1 / json
aws --profile local --endpoint-url=http://localhost:4566 s3 ls

awslocal wrapper (from source)

chmod +x bin/awslocal
./bin/awslocal s3 ls
./bin/awslocal dynamodb list-tables

---

boto3 Usage Patterns

Universal client factory

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

s3 = aws_client("s3")

# Create bucket and upload
s3.create_bucket(Bucket="my-bucket")
s3.put_object(Bucket="my-bucket", Key="hello.txt", Body=b"Hello, MiniStack!")

# Download
obj = s3.get_object(Bucket="my-bucket", Key="hello.txt")
print(obj["Body"].read())  # b'Hello, MiniStack!'

# List objects
response = s3.list_objects_v2(Bucket="my-bucket")
for item in response.get("Contents", []):
    print(item["Key"])

# Copy object
s3.copy_object(
    Bucket="my-bucket",
    CopySource={"Bucket": "my-bucket", "Key": "hello.txt"},
    Key="hello-copy.txt",
)

# Enable versioning
s3.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

sqs = aws_client("sqs")

# Standard queue
queue = 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 queue
fifo = sqs.create_queue(
    QueueName="my-queue.fifo",
    Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"},
)

# Dead-letter queue setup
dlq = 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

import json
ddb = aws_client("dynamodb")

# Create table
ddb.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 / Delete
ddb.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

# Query
result = ddb.query(
    TableName="Users",
    KeyConditionExpression="userId = :uid",
    ExpressionAttributeValues={":uid": {"S": "u1"}},
)

# Batch write
ddb.batch_write_item(
    RequestItems={
        "Users": [
            {"PutRequest": {"Item": {"userId": {"S": "u2"}, "createdAt": {"N": "1700000001"}, "name": {"S": "Bob"}}}},
        ]
    }
)

# TTL
ddb.update_time_to_live(
    TableName="Users",
    TimeToLiveSpecification={"Enabled": True, "AttributeName": "expiresAt"},
)

SNS + SQS Fanout

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 queues
sns.publish(TopicArn=topic_arn, Message="hello fanout", Subject="test")

Lambda

import zipfile, io

# Create a zip with handler code
buf = 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 synchronously
import json
response = 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 mapping
lam.create_event_source_mapping(
    EventSourceArn=queue_arn,
    FunctionName="my-function",
    BatchSize=10,
    Enabled=True,
)

SecretsManager

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

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 path
params = ssm.get_parameters_by_path(Path="/app/", Recursive=True)
for p in params["Parameters"]:
    print(p["Name"], p["Value"])

Kinesis

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 records
shards = 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

eb = aws_client("events")

# Create a custom bus
eb.create_event_bus(Name="my-bus")

# Put a rule targeting a Lambda
eb.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

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, -exclusions
filtered = logs.filter_log_events(
    logGroupName="/app/service",
    filterPattern="Request*",
)

---

Testing Patterns

pytest fixture (recommended)

import pytest
import boto3

MINISTACK_ENDPOINT = "http://localhost:4566"

@pytest.fixture(scope="session")
def aws_endpoint():
    return MINISTACK_ENDPOINT

@pytest.fixture
def 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.fixture
def 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

# .github/workflows/test.yml
name: 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:4566

Using AWS_ENDPOINT_URL env var (boto3 >= 1.28)

import os
import boto3

# If AWS_ENDPOINT_URL is set, boto3 uses it automatically — no endpoint_url kwarg needed
# export AWS_ENDPOINT_URL=http://localhost:4566
s3 = boto3.client("s3")  # picks up AWS_ENDPOINT_URL automatically

---

Supported Services (25+)

---

Troubleshooting

Connection refused on port 4566

# Check if ministack is running
curl http://localhost:4566/_localstack/health
# Start it
ministack
# or
docker run -p 4566:4566 nahuelnucera/ministack

`NoCredentialsError` from boto3

export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1
# Any non-empty values work — MiniStack doesn't validate credentials

`InvalidSignatureException`

• 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

# Enable persistence
S3_PERSIST=1 ministack
# or in Docker
docker run -p 4566:4566 -e S3_PERSIST=1 -v $(pwd)/data:/data nahuelnucera/ministack

Port conflict

GATEWAY_PORT=5000 ministack
# Then use http://localhost:5000 as endpoint

Migrating from LocalStack

• Replace all http://localhost:4566 endpoint URLs — they stay the same.
• Remove LOCALSTACK_AUTH_TOKEN / LOCALSTACK_API_KEY env vars (not needed).
• Replace localstack/localstack Docker image with nahuelnucera/ministack.
• All boto3 client code works without modification.