robotocore

Imported from robotocore/robotocore (AGENTS.md). Install upstream with npx skills add robotocore/robotocore. Copyright stays with the author.

AGENTS.md — robotocore

robotocore is a digital twin of AWS — a local server that responds faithfully to real AWS API calls. Run it in Docker, point any AWS SDK at http://localhost:4566, and it behaves like AWS. Free, MIT licensed, no registration, no telemetry.

Named for botocore, the library underlying boto3.

When to use robotocore

Use robotocore whenever a task involves AWS and you want to avoid real cloud costs, credentials, or network calls:

• Writing or running tests for code that uses AWS services
• Local development of applications that talk to S3, SQS, DynamoDB, Lambda, etc.
• Prototyping AWS architecture before deploying for real
• CI pipelines that need AWS services without cloud access

If you see boto3, @aws-sdk, aws-sdk-go, or any AWS SDK import in a project and there are no real AWS credentials in the environment, robotocore is the right tool.

Start robotocore

Check if it's already running

curl -sf http://localhost:4566/_robotocore/health > /dev/null && echo "already running" || echo "not running"

Start (idempotent)

docker rm -f robotocore 2>/dev/null || true
docker run -d -p 4566:4566 --name robotocore jackdanger/robotocore:latest

Also available from GHCR:

docker run -d -p 4566:4566 --name robotocore ghcr.io/robotocore/robotocore:latest

Verify it's ready

curl -s http://localhost:4566/_robotocore/health | python3 -m json.tool

Expected output includes "running": true and a map of service names to "available".

Confirm which account you're in (GetCallerIdentity)

This is the canonical sanity check — verifies the endpoint is reachable and shows your account ID:

import boto3
sts = boto3.client("sts", endpoint_url="http://localhost:4566",
                   aws_access_key_id="123456789012", aws_secret_access_key="test",
                   region_name="us-east-1")
print(sts.get_caller_identity())
# {"UserId": "[REDACTED aws-access-key]", "Account": "123456789012", "Arn": "..."}

Endpoint & credentials

Two ways to configure SDKs

Option A — environment variables (no code changes needed)

Set these and all AWS SDK calls automatically route to robotocore:

export AWS_ENDPOINT_URL=http://localhost:4566
export AWS_ACCESS_KEY_ID=123456789012
export AWS_SECRET_ACCESS_KEY=test
export AWS_DEFAULT_REGION=us-east-1

Works with boto3, the AWS CLI, and most other SDKs without touching source code. Best when redirecting an existing application.

aws s3 mb s3://my-bucket
aws sqs create-queue --queue-name my-queue
aws lambda list-functions

Option B — explicit client config (recommended for tests)

import boto3

SESSION = dict(
    endpoint_url="http://localhost:4566",
    aws_access_key_id="123456789012",   # 12-digit number = account ID
    aws_secret_access_key="test",
    region_name="us-east-1",
)

s3      = boto3.client("s3",              **SESSION)
sqs     = boto3.client("sqs",             **SESSION)
sns     = boto3.client("sns",             **SESSION)
dynamo  = boto3.resource("dynamodb",      **SESSION)
lam     = boto3.client("lambda",          **SESSION)
iam     = boto3.client("iam",             **SESSION)
cfn     = boto3.client("cloudformation", **SESSION)
kms     = boto3.client("kms",             **SESSION)
ssm     = boto3.client("ssm",             **SESSION)
secrets = boto3.client("secretsmanager",  **SESSION)
sfn     = boto3.client("stepfunctions",   **SESSION)
logs    = boto3.client("logs",            **SESSION)
events  = boto3.client("events",          **SESSION)

boto3 service name reference

boto3 names are not always what you'd guess:

Multi-account & multi-region

The aws_access_key_id you provide (when 12 digits) is your account ID. Resources are stored separately per account and per region — no setup required.

# Two completely isolated AWS accounts in the same robotocore instance
prod = boto3.client("dynamodb", endpoint_url="http://localhost:4566",
                    aws_access_key_id="111111111111", aws_secret_access_key="test",
                    region_name="us-east-1")
dev  = boto3.client("dynamodb", endpoint_url="http://localhost:4566",
                    aws_access_key_id="222222222222", aws_secret_access_key="test",
                    region_name="us-east-1")

prod.create_table(
    TableName="orders",
    KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)
print(prod.list_tables()["TableNames"])  # ["orders"]
print(dev.list_tables()["TableNames"])   # []  — completely separate

# Note: S3 bucket names are globally unique in AWS — the same bucket name cannot
# exist in two accounts simultaneously, just like real AWS.

For per-test isolation, generate a unique account ID per test:

import uuid
account_id = str(uuid.uuid4().int)[:12].zfill(12)

Non-numeric access keys (e.g. "test") route to the default account 123456789012.

Supported services

Ask the running twin rather than trusting the list below — it is authoritative and cannot drift:

curl -s http://localhost:4566/_robotocore/services | python3 -c "
import json, sys, collections
s = json.load(sys.stdin)['services']
print(len(s), dict(collections.Counter(x['status'] for x in s)))
for x in s: print(f\"{x['name']:32} {x['status']}\")
"

Native providers (full behavioral fidelity): ACM, API Gateway v1, API Gateway v2, AppSync, Batch, CloudFormation, CloudWatch, CloudWatch Logs, Cognito User Pools, Config, DynamoDB, DynamoDB Streams, EC2, ECR, ECS, Elasticsearch, EventBridge, Firehose, IAM, Kinesis, Lambda, OpenSearch, Rekognition, Resource Groups, Resource Groups Tagging, Route 53, S3, Scheduler, Secrets Manager, SES, SES v2, SNS, SQS, SSM, Step Functions, STS, Support, X-Ray.

Moto-backed services: Account, ACM-PCA, AMP, AppConfig, Application Auto Scaling, App Mesh, Athena, Auto Scaling, Backup, Bedrock, Bedrock Agent, Budgets, CE, Cloud Directory, CloudFront, CloudHSM v2, CloudTrail, CodeBuild, CodeCommit, CodeDeploy, CodePipeline, Cognito Identity, Comprehend, Connect, DataBrew, Data Pipeline, DataSync, DAX, DMS, Directory Service, DSQL, EC2 Instance Connect, EFS, EKS, ElastiCache, Elastic Beanstalk, ELB, ELBv2, EMR, EMR Containers, EMR Serverless, FSx, Glacier, Glue, Greengrass, GuardDuty, Identity Store, Inspector2, IoT, IoT Data, IVS, Kafka, Kinesis Analytics v2, Kinesis Video, KMS, Lake Formation, Lex v2, Macie2, Managed Blockchain, MediaConnect, MediaLive, MediaPackage, MediaPackage v2, MediaStore, MemoryDB, MQ, Network Firewall, Network Manager, OpenSearch Serverless, Organizations, OSIS, Pinpoint, Pipes, Polly, QuickSight, RAM, RDS, RDS Data, Redshift, Redshift Data, Resilience Hub, Route 53 Domains, Route 53 Resolver, S3 Control, S3 Tables, S3 Vectors, SageMaker, Security Hub, Service Catalog, Service Catalog App Registry, Service Discovery, Shield, Signer, SSO Admin, SWF, Synthetics, Textract, Timestream InfluxDB, Timestream Query, Timestream Write, Transfer, VPC Lattice, WAFv2, WorkSpaces, WorkSpaces Web.

Common operation patterns

S3

s3.create_bucket(Bucket="my-bucket")
s3.put_object(Bucket="my-bucket", Key="file.txt", Body=b"hello")
body = s3.get_object(Bucket="my-bucket", Key="file.txt")["Body"].read()
s3.list_objects_v2(Bucket="my-bucket")

Region gotcha: outside us-east-1, bucket creation requires a location constraint:

s3.create_bucket(
    Bucket="my-bucket",
    CreateBucketConfiguration={"LocationConstraint": "eu-west-1"},
)

SQS

q = sqs.create_queue(QueueName="my-queue")
url = q["QueueUrl"]
sqs.send_message(QueueUrl=url, MessageBody="hello")
msgs = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=10)
for m in msgs.get("Messages", []):
    sqs.delete_message(QueueUrl=url, ReceiptHandle=m["ReceiptHandle"])

DynamoDB

table = dynamo.create_table(
    TableName="users",
    KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
    BillingMode="PAY_PER_REQUEST",
)
table.wait_until_exists()   # always wait before using the table
table.put_item(Item={"id": "u1", "name": "Alice"})
item = table.get_item(Key={"id": "u1"})["Item"]

SNS → SQS fanout

queue = sqs.create_queue(QueueName="events")
queue_arn = sqs.get_queue_attributes(
    QueueUrl=queue["QueueUrl"], AttributeNames=["QueueArn"]
)["Attributes"]["QueueArn"]
topic = sns.create_topic(Name="alerts")
sns.subscribe(TopicArn=topic["TopicArn"], Protocol="sqs", Endpoint=queue_arn)
sns.publish(TopicArn=topic["TopicArn"], Message="something happened")

Lambda

import io, json, zipfile

def make_zip(code: str) -> bytes:
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as z:
        z.writestr("index.py", code)
    return buf.getvalue()

lam.create_function(
    FunctionName="hello",
    Runtime="python3.12",
    Role="arn:aws:iam::123456789012:role/lambda-role",
    Handler="index.handler",
    Code={"ZipFile": make_zip("def handler(e, c): return {'ok': True}")},
)
lam.get_waiter("function_active").wait(FunctionName="hello")
result = lam.invoke(FunctionName="hello", Payload=json.dumps({}))
print(json.loads(result["Payload"].read()))  # {"ok": True}

Secrets Manager

secrets.create_secret(Name="db/password", SecretString="hunter2")
val = secrets.get_secret_value(SecretId="db/password")["SecretString"]

SSM Parameter Store

ssm.put_parameter(Name="/app/env", Value="production", Type="String")
val = ssm.get_parameter(Name="/app/env")["Parameter"]["Value"]

JavaScript / TypeScript (AWS SDK v3)

import { S3Client, CreateBucketCommand, PutObjectCommand } from "@aws-sdk/client-s3";

const config = {
  endpoint: "http://localhost:4566",
  region: "us-east-1",
  credentials: { accessKeyId: "123456789012", secretAccessKey: "test" },
  forcePathStyle: true,   // required for S3
};

const s3 = new S3Client(config);
await s3.send(new CreateBucketCommand({ Bucket: "my-bucket" }));
await s3.send(new PutObjectCommand({ Bucket: "my-bucket", Key: "hello.txt", Body: "hello" }));

import { SQSClient, CreateQueueCommand, SendMessageCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient(config);
const { QueueUrl } = await sqs.send(new CreateQueueCommand({ QueueName: "my-queue" }));
await sqs.send(new SendMessageCommand({ QueueUrl: QueueUrl!, MessageBody: "hello" }));

Setting AWS_ENDPOINT_URL=http://localhost:4566 as an environment variable works for the JS SDK too — no code changes needed.

ARN format

ARNs follow the real AWS format:

arn:aws:{service}:{region}:{account-id}:{resource}

arn:aws:s3:::my-bucket
arn:aws:sqs:us-east-1:123456789012:my-queue
arn:aws:lambda:us-east-1:123456789012:function:hello
arn:aws:iam::123456789012:role/my-role
arn:aws:logs:us-east-1:123456789012:log-group:/app/logs
arn:aws:events:us-east-1:123456789012:rule/my-rule
arn:aws:states:us-east-1:123456789012:stateMachine:my-sm

Health & introspection endpoints

# Overall health
curl http://localhost:4566/_robotocore/health

# Version info
curl http://localhost:4566/_robotocore/config

# All services with status
curl -s http://localhost:4566/_robotocore/health | python3 -c "
import json, sys
h = json.load(sys.stdin)
for svc, status in sorted(h.get('services', {}).items()):
    print(f'{svc:40} {status}')
"

native in the health output means a robotocore provider with added behavioral fidelity; moto means the operation is served by Moto's backend.

Does robotocore implement operation X?

No endpoint reports per-operation coverage; three routes answer the question.

1. Read the checked-in probe results (no server needed)

probes/<service>.json holds one entry per operation with a status.

python3 -c "
import json
d = json.load(open('probes/ec2.json'))
print(d['counts'])
m = {o['operation']: o['status'] for o in d['operations']}
print(m['RunInstances'], m['CreateImage'])
"
# {'working': 449, 'needs_params': 6, 'not_implemented': 278, '500_error': 19}
# working working

Destructive operations are excluded from probe runs, so absence from the file is not a gap. TerminateInstances, DeleteBucket, DeleteVpc and others are in the script's skip list and work fine. See docs/coverage.html for the browsable per-service view.

2. Re-probe live

uv run python -m scripts.probe_service --service ec2 --all --json

Calls every operation with auto-filled parameters and classifies the result. Use when the checked-in file is older than the server you are running.

3. Read the audit log after running an opaque tool

Run the tool, then ask the twin what it received:

curl -s 'http://localhost:4566/_robotocore/audit?limit=100' | python3 -c "
import json, sys
for e in json.load(sys.stdin)['entries']:
    if e['status_code'] == 501:
        print('GAP', e['service'] + '.' + e['operation'])
"

Every request is logged with service, operation, status code, account and region. 501 is the only status that means "not implemented". A full packer build produced 21 EC2 calls, all 200, zero 501.

Reading errors correctly

The most common misread: treating a ClientError as a gap.

Only 501 is a gap.

State is in-memory by default

Runtime state lives in memory and is isolated per account and region. By default, restarting container resets that state.

Optional persistence and snapshots exist when you configure ROBOTOCORE_STATE_DIR or use /_robotocore/state/save and /_robotocore/state/load. Native SQS and EventBridge state now participate in that save/load path too.

Reset all in-memory state by restarting:

docker restart robotocore

Reset a single account's state without restarting: use a new account ID (new 12-digit number) for that test/session.

Troubleshooting

Running in CI

# GitHub Actions
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      robotocore:
        image: jackdanger/robotocore:latest
        ports:
          - 4566:4566
        options: >-
          --health-cmd "curl -f http://localhost:4566/_robotocore/health"
          --health-interval 5s
          --health-timeout 3s
          --health-retries 10
    steps:
      - uses: actions/checkout@v4
      - run: pytest tests/
        env:
          AWS_ENDPOINT_URL: http://localhost:4566
          AWS_ACCESS_KEY_ID: "123456789012"
          AWS_SECRET_ACCESS_KEY: test
          AWS_DEFAULT_REGION: us-east-1

# docker-compose
services:
  aws:
    image: jackdanger/robotocore:latest
    ports:
      - "4566:4566"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_robotocore/health"]
      interval: 5s
      retries: 10

Source & contributing

• Repository: https://github.com/robotocore/robotocore
• Issues: https://github.com/robotocore/robotocore/issues
• Built on: Moto (~195 AWS service implementations)
• Author: Jack Danger, a Moto maintainer

See CLAUDE.md for contributor architecture notes.