Python Aurora Agent Skill

AWS Aurora PostgreSQL and Aurora DSQL Python integration using asyncpg. Use when building Python backends with Aurora PostgreSQL or Aurora DSQL.

1k tokens

context cost

the whole folder, loaded on every use

files

instructions only

copies elsewhere

how many repositories repackaged it

stars on the repo

on the repository, not the skill itself

Install

one command, takes just this skill from the repository

npx skills add https://github.com/vercel-labs/nextjs-fastapi-chat-app-starter --skill python-aurora

The instruction itself

7 sections, as written by the author

AWS Aurora Python Integration

Use asyncpg for queries, boto3 for IAM auth token generation, and the Vercel Python SDK for OIDC token retrieval. On Vercel, AWS credentials are obtained via OIDC federation: call get_vercel_oidc_token() from vercel.oidc.aio, then exchange that token for temporary AWS credentials via sts.assume_role_with_web_identity(). Do not rely on VERCEL_OIDC_TOKEN as an environment variable; it is not available in production runtime.

Guidelines

Use asyncpg.create_pool() with the IAM token as the password. Async access avoids blocking FastAPI's event loop.
Authenticate via OIDC: call get_vercel_oidc_token() to get the Vercel OIDC token, pass it to sts.assume_role_with_web_identity() to get temporary credentials, then use those credentials with rds.generate_db_auth_token().
Use $1, $2 parameter syntax for queries (asyncpg's native format). Never use f-strings for SQL.
Enable SSL (ssl="require") — IAM auth requires an encrypted connection.
Wrap route handlers in try/except and return clear error JSON. Unhandled errors get buried in deep ASGI stack traces and could lead you in the wrong direction.

Aurora DSQL Restrictions

If the target is Aurora DSQL (not standard Aurora PostgreSQL), these PostgreSQL features are not available:

No SERIAL / BIGSERIAL / SMALLSERIAL — use VARCHAR IDs (e.g. nanoid)
No temporary tables, triggers, sequences, partitions
No ON UPDATE clause in CREATE TABLE
Must COMMIT after each DDL statement
Must use CREATE INDEX ASYNC (no ASC/DESC in index ordering)

Dependencies

asyncpg: async PostgreSQL driver
boto3: AWS SDK for OIDC credential exchange and IAM token generation
vercel: Vercel Python SDK for runtime OIDC token retrieval

Environment Variables

PGHOST — Aurora cluster endpoint
PGUSER — database user (default: postgres)
PGDATABASE — database name (default: postgres)
AWS_REGION — AWS region
AWS_ROLE_ARN — IAM role to assume via OIDC

Setup

import os
import asyncpg
import boto3
from vercel.oidc import decode_oidc_payload
from vercel.oidc.aio import get_vercel_oidc_token


async def _get_aws_credentials() -> dict:
    """Exchange Vercel's OIDC token for temporary AWS credentials."""
    token = await get_vercel_oidc_token()
    payload = decode_oidc_payload(token)
    project_id = payload.get("project_id")

    sts = boto3.client("sts", region_name=os.environ["AWS_REGION"])
    resp = sts.assume_role_with_web_identity(
        RoleArn=os.environ["AWS_ROLE_ARN"],
        RoleSessionName=f"aurora-fastapi-{project_id or 'session'}",
        WebIdentityToken=token,
    )
    return resp["Credentials"]


async def generate_auth_token() -> str:
    creds = await _get_aws_credentials()
    client = boto3.client(
        "rds",
        region_name=os.environ["AWS_REGION"],
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
    )
    return client.generate_db_auth_token(
        DBHostname=os.environ["PGHOST"],
        Port=5432,
        DBUsername=os.environ.get("PGUSER", "postgres"),
    )


pool: asyncpg.Pool | None = None


async def get_pool() -> asyncpg.Pool:
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(
            host=os.environ["PGHOST"],
            port=5432,
            user=os.environ.get("PGUSER", "postgres"),
            password=await generate_auth_token(),
            database=os.environ.get("PGDATABASE", "postgres"),
            ssl="require",
        )
    return pool

Usage

# Error handling omitted for brevity — see Guidelines
from fastapi import FastAPI

app = FastAPI()

@app.get("/todos")
async def list_todos():
    p = await get_pool()
    rows = await p.fetch("SELECT * FROM todos ORDER BY created_at DESC")
    return [dict(r) for r in rows]

@app.post("/todos")
async def create_todo(title: str):
    p = await get_pool()
    row = await p.fetchrow("INSERT INTO todos (title) VALUES ($1) RETURNING *", title)
    return dict(row)

@app.patch("/todos/{todo_id}")
async def update_todo(todo_id: int, completed: bool):
    p = await get_pool()
    row = await p.fetchrow("UPDATE todos SET completed = $1 WHERE id = $2 RETURNING *", completed, todo_id)
    return dict(row)

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    p = await get_pool()
    await p.execute("DELETE FROM todos WHERE id = $1", todo_id)

How to use it

Copy the folder

Take vercel-labs/python-aurora from the repository into ~/.claude/skills for personal use, or into .claude/skills inside a project.

Check the name does not clash

The agent identifies a skill by the name field in its header. Two skills with the same name cannot sit side by side — one of them will be ignored.