### Quick Start

Source: https://context7.com/nahuel990/ministack/llms.txt

Instructions on how to install and run MiniStack locally using pip, Docker, or Docker Compose.

```APIDOC
## Quick Start

Starting MiniStack locally to emulate AWS services on port 4566.

```bash
# Option 1: Install via PyPI
pip install ministack
ministack

# Option 2: Run via Docker
docker run -p 4566:4566 nahuelnucera/ministack

# Option 3: Docker Compose
git clone https://github.com/Nahuel990/ministack
cd ministack
docker compose up -d

# Verify MiniStack is running
curl http://localhost:4566/_localstack/health
```
```

--------------------------------

### Install and Run MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Instructions for installing and running MiniStack using different methods: PyPI, Docker Hub, or cloning the repository. It also includes a command to verify the running instance.

```bash
pip install ministack
ministack
# Runs on http://localhost:4566 — use GATEWAY_PORT=XXXX to change
```

```bash
docker run -p 4566:4566 nahuelnucera/ministack
```

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

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

--------------------------------

### Initialize MiniStack Test Environment

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Installs the necessary Python dependencies and starts the MiniStack service using Docker Compose for testing purposes.

```bash
# Install test dependencies
pip install boto3 pytest duckdb docker cbor2

# Start MiniStack
docker compose up -d
```

--------------------------------

### Start MiniStack Locally

Source: https://context7.com/nahuel990/ministack/llms.txt

Methods to initialize the MiniStack emulator using Python, Docker, or Docker Compose. Includes a health check command to verify the service is operational.

```bash
pip install ministack
ministack

docker run -p 4566:4566 nahuelnucera/ministack

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

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

--------------------------------

### AWS CLI with MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Demonstrates how to use the AWS Command Line Interface (CLI) with MiniStack. It shows two methods: using environment variables for access keys and region, or configuring a named profile. Examples include creating S3 buckets, SQS queues, and interacting with DynamoDB and STS.

```bash
# Option A — environment variables (no profile needed)
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1

aws --endpoint-url=http://localhost:4566 s3 mb s3://my-bucket
aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queue
aws --endpoint-url=http://localhost:4566 dynamodb list-tables
aws --endpoint-url=http://localhost:4566 sts get-caller-identity
```

```bash
# Option B — named profile (must pass --profile on every command)
aws configure --profile local
# AWS Access Key ID: test
# AWS Secret Access Key: test
# Default region: us-east-1
# Default output format: json

aws --profile local --endpoint-url=http://localhost:4566 s3 mb s3://my-bucket
aws --profile local --endpoint-url=http://localhost:4566 s3 cp ./file.txt s3://my-bucket/
aws --profile local --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queue
aws --profile local --endpoint-url=http://localhost:4566 dynamodb list-tables
aws --profile local --endpoint-url=http://localhost:4566 sts get-caller-identity
```

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

--------------------------------

### Run MiniStack Tests Locally

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This is a set of bash commands for setting up and running tests locally for the MiniStack project. It includes starting the Docker compose environment, installing dependencies, and executing pytest.

```bash
# Start the stack
docker compose up -d

# Install test dependencies
pip install boto3 pytest duckdb docker cbor2

# Run all tests
pytest tests/ -v

# Run a specific service
pytest tests/ -v -k "cognito"
```

--------------------------------

### Run MiniStack via Docker

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Starts the MiniStack container with state persistence enabled. This configuration maps local directories to the container to ensure data survives restarts.

```bash
docker run -p 4566:4566 \
  -e PERSIST_STATE=1 \
  -e STATE_DIR=/data/ministack-state \
  -v /tmp/ministack-data:/data \
  nahuelnucera/ministack
```

--------------------------------

### Boto3 Integration with MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Shows how to use the AWS SDK for Python (boto3) with MiniStack. It provides a helper function to create clients for various services, all pointing to the MiniStack endpoint. Examples include S3 bucket creation and object manipulation, SQS queue operations, DynamoDB table creation and item insertion, SSM parameter management, Secrets Manager secret creation, and Kinesis stream operations.

```python
import boto3

# All clients use the same endpoint
def client(service):
    return boto3.client(
        service,
        endpoint_url="http://localhost:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        region_name="us-east-1",
    )

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

# SQS
sqs = client("sqs")
q = sqs.create_queue(QueueName="my-queue")
sqs.send_message(QueueUrl=q["QueueUrl"], MessageBody="hello")
msgs = sqs.receive_message(QueueUrl=q["QueueUrl"])
print(msgs["Messages"][0]["Body"])  # hello

# DynamoDB
ddb = client("dynamodb")
ddb.create_table(
    TableName="Users",
    KeySchema=[{"AttributeName": "userId", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "userId", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)
ddb.put_item(TableName="Users", Item={"userId": {"S": "u1"}, "name": {"S": "Alice"}})

# SSM Parameter Store
ssm = client("ssm")
ssm.put_parameter(Name="/app/db/host", Value="localhost", Type="String")
param = ssm.get_parameter(Name="/app/db/host")
print(param["Parameter"]["Value"])  # localhost

# Secrets Manager
sm = client("secretsmanager")
sm.create_secret(Name="db-password", SecretString='{"password":"s3cr3t"}')

# Kinesis
kin = client("kinesis")
kin.create_stream(StreamName="events", ShardCount=1)
kin.put_record(StreamName="events", Data=b'{"event":"click"}', PartitionKey="user1")
```

--------------------------------

### Register ECS Task Definition

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Provides an example of creating an ECS cluster and registering a task definition with container configurations, such as image and port mapping.

```python
import boto3

ecs = boto3.client("ecs", endpoint_url="http://localhost:4566",
                   aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1")

ecs.create_cluster(clusterName="dev")

ecs.register_task_definition(
    family="web",
    containerDefinitions=[{
        "name": "nginx",
        "image": "nginx:alpine",
        "cpu": 128,
        "memory": 256,
        "portMappings": [{"containerPort": 80, "hostPort": 8080}],
    }],
)
```

--------------------------------

### Manage ECS Tasks with Boto3

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Demonstrates how to programmatically start and stop ECS tasks using the AWS SDK for Python (boto3). This is useful for automating containerized service lifecycles within the MiniStack environment.

```python
resp = ecs.run_task(cluster="dev", taskDefinition="web", count=1)
task_arn = resp["tasks"][0]["taskArn"]

# Stop it (removes the container)
ecs.stop_task(cluster="dev", task=task_arn)
```

--------------------------------

### Execute Athena SQL Queries via DuckDB

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Illustrates running SQL queries through the Athena interface, which uses DuckDB under the hood. The process involves starting an execution and polling for completion.

```python
import boto3, time

athena = boto3.client("athena", endpoint_url="http://localhost:4566",
                      aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1")

resp = athena.start_query_execution(
    QueryString="SELECT 42 AS answer, 'hello' AS greeting",
    ResultConfiguration={"OutputLocation": "s3://athena-results/"},
)
query_id = resp["QueryExecutionId"]

while True:
    status = athena.get_query_execution(QueryExecutionId=query_id)
    if status["QueryExecution"]["Status"]["State"] == "SUCCEEDED":
        break
    time.sleep(0.1)

results = athena.get_query_results(QueryExecutionId=query_id)
for row in results["ResultSet"]["Rows"][1:]:
    print([col["VarCharValue"] for col in row["Data"]])
```

--------------------------------

### Publish Event to AWS EventBridge

Source: https://github.com/nahuel990/ministack/blob/master/README.md

This snippet shows how to publish an event to AWS EventBridge using the 'events' client. It specifies the source, detail type, event details, and the event bus name. Ensure Boto3 is installed and configured.

```python
import boto3

client = boto3.client('events')
client.put_events(Entries=[
    {
        "Source": "myapp",
        "DetailType": "UserSignup",
        "Detail": '{"userId": "123"}',
        "EventBusName": "default"
    },
])
```

--------------------------------

### Connect to PostgreSQL Instance with psycopg2

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates how to establish a connection to a PostgreSQL database, create a table, insert data, and query the results using the psycopg2 library.

```python
import psycopg2

conn = psycopg2.connect(
    host=endpoint["Address"],
    port=endpoint["Port"],
    user="admin",
    password="password123",
    dbname="appdb"
)

cursor = conn.cursor()
cursor.execute("CREATE TABLE users (id SERIAL PRIMARY KEY, name VARCHAR(100))")
cursor.execute("INSERT INTO users (name) VALUES ('Alice')")
conn.commit()

cursor.execute("SELECT * FROM users")
print(cursor.fetchall())

conn.close()
```

--------------------------------

### Configure API Gateway HTTP APIs

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates creating an HTTP API, setting up a Lambda proxy integration, and deploying the API using boto3.

```python
import boto3
import requests

apigw = boto3.client(
    "apigatewayv2",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

api = apigw.create_api(Name="my-api", ProtocolType="HTTP")
api_id = api["ApiId"]

integration = apigw.create_integration(
    ApiId=api_id,
    IntegrationType="AWS_PROXY",
    IntegrationUri="arn:aws:lambda:us-east-1:000000000000:function:hello-function",
    PayloadFormatVersion="2.0"
)

apigw.create_route(ApiId=api_id, RouteKey="GET /hello", Target=f"integrations/{integration['IntegrationId']}")
apigw.create_stage(ApiId=api_id, StageName="$default", AutoDeploy=True)

response = requests.get(f"http://{api_id}.execute-api.localhost:4566/hello")
print(response.json())
```

--------------------------------

### Connect to RDS Instance using Boto3 and Psycopg2

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Demonstrates how to create a real RDS instance using the MiniStack endpoint and connect to it using a standard PostgreSQL driver. It highlights that the returned endpoint is a functional database instance.

```python
import boto3
import psycopg2

rds = boto3.client("rds", endpoint_url="http://localhost:4566",
                   aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1")

resp = rds.create_db_instance(
    DBInstanceIdentifier="mydb",
    DBInstanceClass="db.t3.micro",
    Engine="postgres",
    MasterUsername="admin",
    MasterUserPassword="password",
    DBName="appdb",
    AllocatedStorage=20,
)

endpoint = resp["DBInstance"]["Endpoint"]
conn = psycopg2.connect(
    host=endpoint["Address"],
    port=endpoint["Port"],
    user="admin",
    password="password",
    dbname="appdb",
)
```

--------------------------------

### Manage ECS Services with Docker

Source: https://context7.com/nahuel990/ministack/llms.txt

Shows how to create an ECS cluster, register a task definition, and run actual Docker containers via the local MiniStack ECS emulator.

```python
import boto3
ecs = boto3.client("ecs", endpoint_url="http://localhost:4566", aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1")
ecs.create_cluster(clusterName="dev-cluster")
ecs.register_task_definition(
    family="web-app",
    containerDefinitions=[{"name": "nginx", "image": "nginx:alpine", "cpu": 128, "memory": 256, "portMappings": [{"containerPort": 80, "hostPort": 8080, "protocol": "tcp"}], "environment": [{"name": "NGINX_HOST", "value": "localhost"}]}]
)
response = ecs.run_task(cluster="dev-cluster", taskDefinition="web-app", count=1)
task_arn = response["tasks"][0]["taskArn"]
ecs.stop_task(cluster="dev-cluster", task=task_arn, reason="Manual stop")
```

--------------------------------

### RDS Service with Real Database using Python

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates RDS emulation with Python, spinning up actual Postgres/MySQL Docker containers. Shows how to create a PostgreSQL instance and retrieve its connection endpoint. Requires boto3 and Docker.

```python
import boto3
import psycopg2  # pip install psycopg2-binary

rds = boto3.client(
    "rds",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Create PostgreSQL instance (starts real Docker container)
response = rds.create_db_instance(
    DBInstanceIdentifier="mydb",
    DBInstanceClass="db.t3.micro",
    Engine="postgres",
    EngineVersion="15",
    MasterUsername="admin",
    MasterUserPassword="password123",
    DBName="appdb",
    AllocatedStorage=20
)

endpoint = response["DBInstance"]["Endpoint"]
print(f"Database endpoint: {endpoint['Address']}:{endpoint['Port']}")

# Wait for instance to be available, then connect
import time
time.sleep(10)  # Allow container to start

```

--------------------------------

### Interact with S3 via Boto3

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates S3 bucket creation, object uploading, versioning, and multipart uploads using the boto3 Python SDK pointed at MiniStack.

```python
import boto3

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

s3.create_bucket(Bucket="my-bucket")
s3.put_object(Bucket="my-bucket", Key="hello.txt", Body=b"Hello, MiniStack!", ContentType="text/plain")
response = s3.get_object(Bucket="my-bucket", Key="hello.txt")
print(response["Body"].read())
```

--------------------------------

### Lambda Service Operations with Python

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates Lambda emulation with Python using boto3. Covers function creation, synchronous invocation, event source mapping, and function URL generation. Requires boto3 and a running Lambda endpoint.

```python
import boto3
import json
import zipfile
import io

lambda_client = boto3.client(
    "lambda",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Create function code as zip
code = b'''
def handler(event, context):
    name = event.get("name", "World")
    return {"statusCode": 200, "body": f"Hello, {name}!"}
'''
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
    zf.writestr("index.py", code)
zip_bytes = zip_buffer.getvalue()

# Create function
lambda_client.create_function(
    FunctionName="hello-function",
    Runtime="python3.9",
    Role="arn:aws:iam::000000000000:role/lambda-role",
    Handler="index.handler",
    Code={"ZipFile": zip_bytes},
    Timeout=30,
    MemorySize=128,
    Environment={"Variables": {"ENV": "development"}}
)

# Invoke function synchronously
response = lambda_client.invoke(
    FunctionName="hello-function",
    InvocationType="RequestResponse",
    Payload=json.dumps({"name": "MiniStack"})
)
result = json.loads(response["Payload"].read())
print(result)  # {'statusCode': 200, 'body': 'Hello, MiniStack!'}

# Create SQS event source mapping
lambda_client.create_event_source_mapping(
    EventSourceArn="arn:aws:sqs:us-east-1:000000000000:my-queue",
    FunctionName="hello-function",
    BatchSize=10,
    Enabled=True
)

# Create function URL
url_config = lambda_client.create_function_url_config(
    FunctionName="hello-function",
    AuthType="NONE"
)
print(f"Function URL: {url_config['FunctionUrl']}")

```

--------------------------------

### Manage SQS Queues via Boto3

Source: https://context7.com/nahuel990/ministack/llms.txt

Shows how to create standard and FIFO queues, send messages, perform long polling, and configure dead-letter queues using boto3.

```python
import boto3
import json

sqs = boto3.client(
    "sqs",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

response = sqs.create_queue(QueueName="my-queue")
queue_url = response["QueueUrl"]

sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({"event": "user_signup", "user_id": "123"}),
    MessageAttributes={"EventType": {"DataType": "String", "StringValue": "signup"}}
)
```

--------------------------------

### S3 Service Emulation

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates how to use the AWS SDK for Python (boto3) to interact with the emulated S3 service, including bucket creation, object upload/download, versioning, and multipart uploads.

```APIDOC
## S3 Service

S3 emulation with full bucket and object operations including versioning, encryption, lifecycle, multipart uploads, and Object Lock.

```python
import boto3

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Create bucket
s3.create_bucket(Bucket="my-bucket")

# Upload object
s3.put_object(
    Bucket="my-bucket",
    Key="hello.txt",
    Body=b"Hello, MiniStack!",
    ContentType="text/plain"
)

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

# List objects
response = s3.list_objects_v2(Bucket="my-bucket")
for obj in response.get("Contents", []):
    print(f"{obj['Key']}: {obj['Size']} bytes")

# Enable versioning
s3.put_bucket_versioning(
    Bucket="my-bucket",
    VersioningConfiguration={"Status": "Enabled"}
)

# Multipart upload for large files
upload = s3.create_multipart_upload(Bucket="my-bucket", Key="large-file.bin")
part = s3.upload_part(
    Bucket="my-bucket",
    Key="large-file.bin",
    UploadId=upload["UploadId"],
    PartNumber=1,
    Body=b"x" * 5_000_000
)
s3.complete_multipart_upload(
    Bucket="my-bucket",
    Key="large-file.bin",
    UploadId=upload["UploadId"],
    MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": part["ETag"]}]}
)
```
```

--------------------------------

### SQS Service Emulation

Source: https://context7.com/nahuel990/ministack/llms.txt

Shows how to use boto3 to interact with the emulated SQS service, including creating standard and FIFO queues, sending/receiving messages, and configuring dead-letter queues.

```APIDOC
## SQS Service

SQS emulation with standard and FIFO queues, dead-letter queues, message attributes, and long polling support.

```python
import boto3
import json

sqs = boto3.client(
    "sqs",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Create standard queue
response = sqs.create_queue(QueueName="my-queue")
queue_url = response["QueueUrl"]

# Create FIFO queue
fifo_response = sqs.create_queue(
    QueueName="my-queue.fifo",
    Attributes={
        "FifoQueue": "true",
        "ContentBasedDeduplication": "true"
    }
)

# Send message
sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({"event": "user_signup", "user_id": "123"}),
    MessageAttributes={
        "EventType": {"DataType": "String", "StringValue": "signup"}
    }
)

# Receive messages with long polling
messages = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=5,
    MessageAttributeNames=["All"]
)

for msg in messages.get("Messages", []):
    print(f"Body: {msg['Body']}")
    # Process message then delete
    sqs.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=msg["ReceiptHandle"]
    )

# Configure dead-letter queue
sqs.set_queue_attributes(
    QueueUrl=queue_url,
    Attributes={
        "RedrivePolicy": json.dumps({
            "deadLetterTargetArn": "arn:aws:sqs:us-east-1:000000000000:dlq",
            "maxReceiveCount": "3"
        })
    }
)
```
```

--------------------------------

### Execute Athena SQL Queries with Python

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates how to run SQL queries against the local Athena engine using Boto3. It includes polling for query completion and parsing the resulting data rows.

```python
response = athena.start_query_execution(
    QueryString="""
        SELECT
            42 AS answer,
            'hello' AS greeting,
            CURRENT_DATE AS today
    """,
    ResultConfiguration={"OutputLocation": "s3://athena-results/"}
)
query_id = response["QueryExecutionId"]

while True:
    status = athena.get_query_execution(QueryExecutionId=query_id)
    state = status["QueryExecution"]["Status"]["State"]
    if state in ("SUCCEEDED", "FAILED", "CANCELLED"):
        break
    time.sleep(0.5)

if state == "SUCCEEDED":
    results = athena.get_query_results(QueryExecutionId=query_id)
    columns = [col["Name"] for col in results["ResultSet"]["ResultSetMetadata"]["ColumnInfo"]]
    for row in results["ResultSet"]["Rows"][1:]:
        values = [col.get("VarCharValue", "") for col in row["Data"]]
        print(f"Row: {values}")
```

--------------------------------

### Orchestrate Workflows with Step Functions

Source: https://context7.com/nahuel990/ministack/llms.txt

Illustrates defining a state machine with various states (Pass, Choice) and executing it using the boto3 Step Functions client.

```python
import boto3
import json
import time

sfn = boto3.client(
    "stepfunctions",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

definition = {
    "StartAt": "ProcessInput",
    "States": {
        "ProcessInput": {"Type": "Pass", "Parameters": {"processed": True, "input.$": "$"}, "Next": "CheckCondition"},
        "CheckCondition": {"Type": "Choice", "Choices": [{"Variable": "$.value", "NumericGreaterThan": 100, "Next": "HighValue"}], "Default": "LowValue"},
        "HighValue": {"Type": "Pass", "Result": {"category": "high"}, "End": True},
        "LowValue": {"Type": "Pass", "Result": {"category": "low"}, "End": True}
    }
}

response = sfn.create_state_machine(name="my-workflow", definition=json.dumps(definition), roleArn="arn:aws:iam::000000000000:role/StepFunctionsRole")
exec_response = sfn.start_execution(stateMachineArn=response["stateMachineArn"], input=json.dumps({"value": 150}))
time.sleep(1)
result = sfn.describe_execution(executionArn=exec_response["executionArn"])
print(f"Status: {result['status']}")
```

--------------------------------

### Run EC2 Instance

Source: https://github.com/nahuel990/ministack/blob/master/README.md

This Python script uses Boto3 to launch a new EC2 instance. It specifies the AMI ID, instance type, and counts. It then retrieves and prints the ID of the newly created instance. Ensure valid AMI ID and AWS credentials.

```python
import boto3

ec2 = boto3.client('ec2')
reservation = ec2.run_instances(
    ImageId='ami-00000001',
    MinCount=1,
    MaxCount=1,
    InstanceType='t3.micro',
)
instance_id = reservation["Instances"][0]["InstanceId"]
print(instance_id)  # i-xxxxxxxxxxxxxxxxx
```

--------------------------------

### DynamoDB Service Operations with Python

Source: https://context7.com/nahuel990/ministack/llms.txt

Demonstrates DynamoDB emulation with Python using boto3. Covers table creation, item put/get operations, queries, and transactional writes. Requires boto3 and a running DynamoDB endpoint.

```python
import boto3
from boto3.dynamodb.conditions import Key, Attr

dynamodb = boto3.client(
    "dynamodb",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Create table
dynamodb.create_table(
    TableName="Users",
    KeySchema=[
        {"AttributeName": "userId", "KeyType": "HASH"},
        {"AttributeName": "sortKey", "KeyType": "RANGE"}
    ],
    AttributeDefinitions=[
        {"AttributeName": "userId", "AttributeType": "S"},
        {"AttributeName": "sortKey", "AttributeType": "S"}
    ],
    BillingMode="PAY_PER_REQUEST"
)

# Put item
dynamodb.put_item(
    TableName="Users",
    Item={
        "userId": {"S": "user-123"},
        "sortKey": {"S": "profile"},
        "name": {"S": "Alice"},
        "email": {"S": "alice@example.com"},
        "age": {"N": "30"}
    }
)

# Get item
response = dynamodb.get_item(
    TableName="Users",
    Key={
        "userId": {"S": "user-123"},
        "sortKey": {"S": "profile"}
    }
)
print(response.get("Item"))

# Query with key condition
response = dynamodb.query(
    TableName="Users",
    KeyConditionExpression="userId = :uid",
    ExpressionAttributeValues={":uid": {"S": "user-123"}}
)

# Transactional write
dynamodb.transact_write_items(
    TransactItems=[
        {
            "Put": {
                "TableName": "Users",
                "Item": {
                    "userId": {"S": "user-456"},
                    "sortKey": {"S": "profile"},
                    "name": {"S": "Bob"}
                }
            }
        },
        {
            "Update": {
                "TableName": "Users",
                "Key": {"userId": {"S": "user-123"}, "sortKey": {"S": "profile"}},
                "UpdateExpression": "SET #n = :name",
                "ExpressionAttributeNames": {"#n": "name"},
                "ExpressionAttributeValues": {":name": {"S": "Alice Updated"}}
            }
        }
    ]
)

```

--------------------------------

### Configure AWS CLI for MiniStack

Source: https://context7.com/nahuel990/ministack/llms.txt

Setting up environment variables and using the --endpoint-url flag to route AWS CLI commands to the local MiniStack instance.

```bash
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1

aws --endpoint-url=http://localhost:4566 s3 mb s3://my-bucket
aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queue
aws --endpoint-url=http://localhost:4566 dynamodb list-tables
aws --endpoint-url=http://localhost:4566 sts get-caller-identity

chmod +x bin/awslocal
./bin/awslocal s3 ls
```

--------------------------------

### Create New Service File in MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This Python code defines a new service emulator for MiniStack. It includes request handling, operation definitions, and state management. It uses core response utilities and logging.

```python
import json
import logging
from ministack.core.responses import json_response, error_response_json, new_uuid

logger = logging.getLogger("myservice")

ACCOUNT_ID = "000000000000"
REGION = "us-east-1"

_state: dict = {}  # in-memory storage


async def handle_request(method, path, headers, body, query_params):
    target = headers.get("x-amz-target", "")
    action = target.split(".")[-1] if "." in target else ""

    try:
        data = json.loads(body) if body else {}
    except json.JSONDecodeError:
        return error_response_json("SerializationException", "Invalid JSON", 400)

    handlers = {
        "OperationOne": _operation_one,
        "OperationTwo": _operation_two,
    }

    handler = handlers.get(action)
    if not handler:
        return error_response_json("InvalidAction", f"Unknown action: {action}", 400)
    return handler(data)


def _operation_one(data):
    return json_response({"result": "ok"})


def _operation_two(data):
    return json_response({})

def reset():
    _state.clear()
```

--------------------------------

### AWS CLI Configuration

Source: https://context7.com/nahuel990/ministack/llms.txt

How to configure the AWS CLI to point to the MiniStack endpoint for local development.

```APIDOC
## AWS CLI Configuration

Configure AWS CLI to use MiniStack endpoint for local development.

```bash
# Set environment variables (no profile needed)
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1

# Use --endpoint-url with any AWS CLI command
aws --endpoint-url=http://localhost:4566 s3 mb s3://my-bucket
aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name my-queue
aws --endpoint-url=http://localhost:4566 dynamodb list-tables
aws --endpoint-url=http://localhost:4566 sts get-caller-identity

# Alternative: Use awslocal wrapper script
chmod +x bin/awslocal
./bin/awslocal s3 ls
./bin/awslocal dynamodb list-tables
```
```

--------------------------------

### Set MiniStack Environment Variables

Source: https://context7.com/nahuel990/ministack/llms.txt

Lists essential environment variables for configuring MiniStack, including ports, persistence directories, and engine settings.

```bash
export GATEWAY_PORT=4566
export S3_PERSIST=1
export S3_DATA_DIR=/tmp/ministack/s3
export PERSIST_STATE=1
export ATHENA_ENGINE=duckdb
```

--------------------------------

### POST /stepfunctions/create_state_machine

Source: https://context7.com/nahuel990/ministack/llms.txt

Defines and creates a new Step Functions state machine using ASL (Amazon States Language).

```APIDOC
## POST /stepfunctions/create_state_machine

### Description
Registers a new state machine definition for use with the Step Functions interpreter.

### Method
POST

### Endpoint
/stepfunctions/create_state_machine

### Parameters
#### Request Body
- **name** (string) - Required - Name of the state machine
- **definition** (string) - Required - JSON string containing the ASL definition
- **roleArn** (string) - Required - IAM role ARN for execution

### Request Example
{
  "name": "my-workflow",
  "definition": "{\"StartAt\": \"ProcessInput\", ...}",
  "roleArn": "arn:aws:iam::000000000000:role/StepFunctionsRole"
}

### Response
#### Success Response (200)
- **stateMachineArn** (string) - The ARN of the created state machine

#### Response Example
{
  "stateMachineArn": "arn:aws:states:us-east-1:000000000000:stateMachine:my-workflow"
}
```

--------------------------------

### Manage ElastiCache Redis Clusters

Source: https://context7.com/nahuel990/ministack/llms.txt

Shows how to create an ElastiCache Redis cluster using boto3 and interact with the real Redis instance using the redis-py library.

```python
import boto3
import redis

elasticache = boto3.client(
    "elasticache",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

response = elasticache.create_cache_cluster(
    CacheClusterId="my-redis",
    Engine="redis",
    CacheNodeType="cache.t3.micro",
    NumCacheNodes=1
)

node = response["CacheCluster"]["CacheNodes"][0]["Endpoint"]
r = redis.Redis(host=node["Address"], port=node["Port"])
r.set("session:user123", "active")
print(r.get("session:user123"))
```

--------------------------------

### Create VPC and Subnet

Source: https://github.com/nahuel990/ministack/blob/master/README.md

This Python code creates a new Virtual Private Cloud (VPC) with a specified CIDR block and then creates a subnet within that VPC, assigning it an availability zone. It utilizes the 'ec2' client from Boto3.

```python
import boto3

ec2 = boto3.client('ec2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(
    VpcId=vpc["Vpc"]["VpcId"],
    CidrBlock='10.0.1.0/24',
    AvailabilityZone='us-east-1a',
)
```

--------------------------------

### CreateLoadBalancer

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Creates a new Application Load Balancer within the Ministack environment.

```APIDOC
## POST /CreateLoadBalancer

### Description
Creates a new Application Load Balancer to manage traffic routing.

### Method
POST

### Endpoint
/CreateLoadBalancer

### Parameters
#### Request Body
- **Name** (string) - Required - The name of the load balancer.

### Request Example
{
  "Name": "my-load-balancer"
}

### Response
#### Success Response (200)
- **LoadBalancerArn** (string) - The unique identifier for the created load balancer.

#### Response Example
{
  "LoadBalancerArn": "arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/app/my-load-balancer/50dc6c495c0c9188"
}
```

--------------------------------

### Interact with ElastiCache Redis Cluster

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Shows how to provision a Redis cache cluster via the ElastiCache API and perform basic key-value operations using the redis-py client.

```python
import boto3
import redis

ec = boto3.client("elasticache", endpoint_url="http://localhost:4566",
                  aws_access_key_id="test", aws_secret_access_key="test", region_name="us-east-1")

resp = ec.create_cache_cluster(
    CacheClusterId="my-redis",
    Engine="redis",
    CacheNodeType="cache.t3.micro",
    NumCacheNodes=1,
)

node = resp["CacheCluster"]["CacheNodes"][0]["Endpoint"]
r = redis.Redis(host=node["Address"], port=node["Port"])
r.set("key", "value")
print(r.get("key"))
```

--------------------------------

### Reset and Configuration Endpoints

Source: https://context7.com/nahuel990/ministack/llms.txt

Endpoints for resetting MiniStack's state and configuring runtime parameters.

```APIDOC
## POST /_ministack/reset

### Description
Resets all state within MiniStack. This is particularly useful for ensuring a clean state before running tests.

### Method
POST

### Endpoint
http://localhost:4566/_ministack/reset

### Response
#### Success Response (200)
- **reset** (string) - Indicates the status of the reset operation, typically "ok".

### Response Example
```json
{
  "reset": "ok"
}
```

## POST /_ministack/config

### Description
Allows runtime configuration of MiniStack parameters. Accepts a JSON object with key-value pairs for configuration settings.

### Method
POST

### Endpoint
http://localhost:4566/_ministack/config

### Parameters
#### Request Body
- **config_key** (string) - Required - The configuration key to set (e.g., "athena.ATHENA_ENGINE").
- **config_value** (string) - Required - The value to set for the configuration key (e.g., "mock").

### Request Example
```json
{
  "athena.ATHENA_ENGINE": "mock"
}
```

### Response
#### Success Response (200)
(No specific response body is detailed, typically an empty success or status message)

### Response Example
(Assumed success response, actual may vary)
```json
{
  "status": "configured"
}
```
```

--------------------------------

### Run MiniStack Test Suite

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Executes the full test suite across all services using pytest. This ensures that all 30 services are functioning correctly within the local environment.

```bash
pytest tests/ -v
```

--------------------------------

### Configure Infrastructure-as-Code for MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Configures various IaC tools to redirect AWS service requests to the local MiniStack instance running on localhost:4566.

```hcl
provider "aws" {
  region                      = "us-east-1"
  access_key                  = "test"
  secret_key                  = "test"
  skip_credentials_validation = true
  skip_metadata_api_check     = true
  skip_requesting_account_id  = true

  endpoints {
    s3             = "http://localhost:4566"
    sqs            = "http://localhost:4566"
    dynamodb       = "http://localhost:4566"
    lambda         = "http://localhost:4566"
    iam            = "http://localhost:4566"
    sts            = "http://localhost:4566"
    secretsmanager = "http://localhost:4566"
    ssm            = "http://localhost:4566"
    kinesis        = "http://localhost:4566"
    sns            = "http://localhost:4566"
    rds            = "http://localhost:4566"
    ecs            = "http://localhost:4566"
    glue           = "http://localhost:4566"
    athena         = "http://localhost:4566"
    elasticache    = "http://localhost:4566"
    stepfunctions  = "http://localhost:4566"
    cloudwatch     = "http://localhost:4566"
    logs           = "http://localhost:4566"
    events         = "http://localhost:4566"
    ses            = "http://localhost:4566"
    apigateway     = "http://localhost:4566"
    firehose       = "http://localhost:4566"
    route53        = "http://localhost:4566"
    cognitoidp     = "http://localhost:4566"
    cognitoidentity = "http://localhost:4566"
    ec2             = "http://localhost:4566"
    emr             = "http://localhost:4566"
    efs             = "http://localhost:4566"
    elbv2           = "http://localhost:4566"
  }
}
```

```typescript
const app = new cdk.App();
process.env.AWS_ENDPOINT_URL = "http://localhost:4566";
```

```yaml
config:
  aws:endpoints:
    - s3: http://localhost:4566
      dynamodb: http://localhost:4566
```

--------------------------------

### CreateRule

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Creates a routing rule for a specific listener to handle incoming traffic based on conditions.

```APIDOC
## POST /CreateRule

### Description
Defines a rule for a listener to route traffic based on conditions like path-pattern or host-header.

### Method
POST

### Endpoint
/CreateRule

### Parameters
#### Request Body
- **ListenerArn** (string) - Required - The ARN of the listener.
- **Conditions** (array) - Required - List of conditions (path-pattern, host-header, etc.).
- **Actions** (array) - Required - List of actions (forward, redirect, fixed-response).

### Request Example
{
  "ListenerArn": "arn:listener:123",
  "Conditions": [{"Field": "path-pattern", "Values": ["/api/*"]}],
  "Actions": [{"Type": "forward", "TargetGroupArn": "arn:tg:123"}]
}

### Response
#### Success Response (200)
- **RuleArn** (string) - The unique identifier for the created rule.

#### Response Example
{
  "RuleArn": "arn:rule:abc"
}
```

--------------------------------

### Add Service Fixture for Testing in MiniStack

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This Python code defines a pytest fixture for testing a new service in MiniStack. It uses a `make_client` utility to create a client for the 'myservice'.

```python
@pytest.fixture(scope="session")
def mysvc():
    return make_client("myservice")
```

--------------------------------

### POST /apigatewayv2/create_api

Source: https://context7.com/nahuel990/ministack/llms.txt

Creates an HTTP API resource for routing requests to backend integrations.

```APIDOC
## POST /apigatewayv2/create_api

### Description
Initializes a new HTTP API gateway resource.

### Method
POST

### Endpoint
/apigatewayv2/create_api

### Parameters
#### Request Body
- **Name** (string) - Required - Name of the API
- **ProtocolType** (string) - Required - Protocol type (e.g., 'HTTP')

### Request Example
{
  "Name": "my-api",
  "ProtocolType": "HTTP"
}

### Response
#### Success Response (200)
- **ApiId** (string) - The unique identifier for the created API

#### Response Example
{
  "ApiId": "abc123xyz"
}
```

--------------------------------

### Add Service Detection Pattern in MiniStack Router

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This Python code adds routing patterns for a new service in `ministack/core/router.py`. It defines target prefixes for header-based routing and host patterns for host-based routing.

```python
SERVICE_PATTERNS = {
    # ... existing ...
    "myservice": {
        "target_prefixes": ["AWSMyService"],   # for X-Amz-Target routing
        "host_patterns": [r"myservice\."],      # for host-based routing
    },
}
```

--------------------------------

### Create AWS Step Functions State Machine

Source: https://github.com/nahuel990/ministack/blob/master/README.md

This code demonstrates how to create a new state machine in AWS Step Functions. It defines the state machine's name, its Amazon States Language definition, and the IAM role ARN for execution. Requires Boto3.

```python
import boto3

client = boto3.client('stepfunctions')
client.create_state_machine(
    name='my-workflow',
    definition='{"StartAt":"Hello","States":{"Hello":{"Type":"Pass","End":true}}}',
    roleArn='arn:aws:iam::000000000000:role/role',
)
```

--------------------------------

### Write Integration Test for MiniStack Service

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This Python code demonstrates writing an integration test for a MiniStack service using pytest. It calls a service operation via the fixture and asserts the expected response.

```python
def test_myservice_operation_one(mysvc):
    resp = mysvc.operation_one(Param="value")
    assert resp["result"] == "ok"
```

--------------------------------

### Register New Service Handler in MiniStack App

Source: https://github.com/nahuel990/ministack/blob/master/CONTRIBUTING.md

This Python code snippet shows how to register a new service handler in the main application file (`app.py`). It maps a service name to its request handling function.

```python
from ministack.services import myservice

SERVICE_HANDLERS = {
    # ... existing ...
    "myservice": myservice.handle_request,
}
```

--------------------------------

### POST /elasticache/create_cache_cluster

Source: https://context7.com/nahuel990/ministack/llms.txt

Creates a new Redis cache cluster using the ElastiCache emulation service.

```APIDOC
## POST /elasticache/create_cache_cluster

### Description
Creates a new Redis cache cluster, which triggers the startup of a real Redis Docker container in the local environment.

### Method
POST

### Endpoint
/elasticache/create_cache_cluster

### Parameters
#### Request Body
- **CacheClusterId** (string) - Required - Unique identifier for the cluster
- **Engine** (string) - Required - The engine type (e.g., 'redis')
- **CacheNodeType** (string) - Required - The compute and memory capacity of the nodes
- **NumCacheNodes** (integer) - Required - Number of cache nodes to create

### Request Example
{
  "CacheClusterId": "my-redis",
  "Engine": "redis",
  "CacheNodeType": "cache.t3.micro",
  "NumCacheNodes": 1
}

### Response
#### Success Response (200)
- **CacheCluster** (object) - Contains cluster details including CacheNodes and Endpoints

#### Response Example
{
  "CacheCluster": {
    "CacheNodes": [{"Endpoint": {"Address": "localhost", "Port": 6379}}]
  }
}
```

--------------------------------

### Lambda Service Operations

Source: https://github.com/nahuel990/ministack/blob/master/README.md

Serverless function execution service supporting CRUD for functions and event source mappings.

```APIDOC
## Lambda Service Operations

### Description
Manages serverless function lifecycle and execution, including support for SQS event source mappings.

### Method
POST

### Endpoint
/lambda/functions/{functionName}/invocations

### Parameters
#### Path Parameters
- **functionName** (string) - Required - The name of the function to invoke.

### Request Example
{
  "FunctionName": "my-function",
  "Payload": "{\"key\": \"value\"}"
}

### Response
#### Success Response (200)
- **Payload** (string) - The function execution result.
```