MCP server on AgentCore Runtime

def ensure_local_db() -> Path:
    """The local database path, downloaded first when this microVM has no copy yet."""
    if not LOCAL_DB.exists():
        download_db()
    return LOCAL_DB


def connect_read_only(path: Path) -> sqlite3.Connection:
    """A connection that can never write: opened read-only and with `query_only` on (covers ATTACHed files too)."""
    conn = sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True)
    conn.execute("PRAGMA query_only = ON")
    conn.row_factory = sqlite3.Row
    return conn


def run_sql(path: Path, query: str, params: tuple = ()) -> dict:
    """Run one statement read-only and return its rows as {"data": [...]}."""
    with closing(connect_read_only(path)) as conn:
        return {"data": [dict(row) for row in conn.execute(query, params).fetchall()]}


def execute(query: str, params: tuple = ()) -> str:
    """Run `query` (with optional `?` parameters) against the local copy; returns the tool's text result."""
    try:
        return json.dumps(run_sql(ensure_local_db(), query, params), default=str)
    except sqlite3.Error as error:
        if "readonly" in str(error) or "query_only" in str(error):
            return READ_ONLY_MESSAGE
        return f"Error executing query: {error}"


@mcp.tool()
def query_db(query: str) -> str:
    """Execute a read-only SQL query (SQLite dialect) on the DataStream Corp database and return the rows as JSON."""
    with lock:
        return execute(query)


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, stateless_http=True)

uv run bootcamp.py deploy
uv run bootcamp.py test --only 1

[CHALLENGE] SKIP stage 1 list_tables tool: no list_tables tool on your MCP server yet
# after your change:
[CHALLENGE] PASS stage 1 list_tables tool: list_tables returned ['audit_log', 'departments', 'employees', 'helpdesk_tickets', 'project_assignments', 'projects', 'sqlite_sequence', 'user_preferences']