Dashboard API: Agent Compatibility Gap Analysis

gistfile1.txt

 ---
  Agent Compatibility Gap Analysis

  ★ Insight ─────────────────────────────────────
  The core blocker is simple: every protected endpoint uses get_current_user_id, which only accepts JWTs. There's no path from a gw_
   API key to a JWT, and no way to create the initial API key without a human in the loop (setup.py requires manual OTP entry). The
  existing get_user_id_from_api_key dependency uses UserKey.secret_key (a different key type), not the gw_ API keys agents would
  use.
  ─────────────────────────────────────────────────

  Gap 1: No API Key → JWT Exchange Endpoint

  Problem: Agents have a gw_ API key but no way to convert it into a JWT. All dashboard endpoints require JWT auth via
  get_current_user_id. The only non-interactive auth path (X-API-Key via get_user_id_from_api_key) uses UserKey.secret_key and is
  only wired to the KYC endpoint.

  Fix: New endpoint POST /v1/auth/token that accepts a gw_ API key and returns a short-lived JWT + refresh token. Flow:
  1. Agent sends X-API-Key: gw_... or Authorization: Bearer gw_...
  2. Server looks up ApiKey by key value, verifies is_active and not expired
  3. Mints JWT with sub=user_id, token_source="api_key", verified_factors=[]
  4. Returns { access_token, refresh_token }
  5. Agent uses JWT for all subsequent calls, refreshes via POST /v1/auth/refresh as usual

  Gap 2: No Headless Bootstrap

  Problem: setup.py (behind make init) requires manual OTP entry — a human must type the email code. No way to create the first API
  key without interactive auth.

  Fix: New script scripts/create_agent.py (behind make init-agent) that:
  1. Creates a User record directly in the DB (no email OTP)
  2. Optionally links a wallet address (public_address)
  3. Creates an ApiKey with permissions=["admin"]
  4. Prints the gw_ key for the agent to use

  This is an admin-only local/CI operation, not an API endpoint.

  Gap 3: gw_ API Keys Not Recognized as Auth

  Problem: The existing get_user_id_from_api_key dependency looks up UserKey.secret_key, not ApiKey.api_key. These are completely
  different tables and key formats. There's no dependency that resolves a gw_ key to a user.

  Fix: This is solved by Gap 1 — the token exchange endpoint handles gw_ key lookup. After exchange, the agent uses the JWT like any
   other client. No need to add gw_ auth to every route.

  Gap 4: Token Claims — verified_factors Requirement

  Problem: Some flows check verified_factors in the JWT (e.g., CLI token creation requires at least one verified factor). Agent
  tokens minted from API key exchange won't have real verified factors.

  Fix: The token exchange endpoint should mint tokens with a special marker (e.g., token_source: "api_key") and empty
  verified_factors. Audit all routes that check verified_factors — currently only POST /v1/authn/cli-token checks this, and agents
  don't need CLI token flow. All other protected routes only check get_current_user_id (which extracts sub, not factors).

  Gap 5: Short Token Lifetime for Agents

  Problem: Current access tokens expire in 30 days, which is fine for dashboard sessions but means agents need to handle refresh.
  Not a blocker, but worth noting.

  Fix: The token exchange endpoint could issue tokens with a configurable or shorter TTL (e.g., 1 hour for agents). Agents use
  refresh tokens to rotate. This is more secure than a 30-day token floating around in agent memory.

  Gap 6: No Agent-Oriented Error Responses

  Problem: Error responses use two inconsistent formats ({"message": "..."} vs {"detail": "..."}). Agents need predictable,
  parseable errors.

  Fix: This is a pre-existing issue, not agent-specific. Low priority — agents can handle both formats. Consider standardizing in a
  future pass.

  ---
  Implementation Plan (Ordered)

  ┌─────┬───────────────────────────────┬──────────────────────────────────────────────────────────────────────────────┬────────┐
  │  #  │             Task              │                                    Files                                     │ Effort │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 1   │ make init-agent script        │ scripts/create_agent.py, Makefile                                            │ Small  │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 2   │ API key → JWT exchange        │ app/api/v1/routes/authn/token.py, app/services/api_key.py,                   │ Medium │
  │     │ endpoint                      │ app/dependencies/auth.py                                                     │        │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 3   │ Audit verified_factors usage  │ Grep + verify no route besides CLI token checks factors                      │ Small  │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 4   │ Agent token TTL               │ app/services/tokens/token_manager.py                                         │ Small  │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 5   │ Tests                         │ tests/unit/, tests/integration/                                              │ Medium │
  ├─────┼───────────────────────────────┼──────────────────────────────────────────────────────────────────────────────┼────────┤
  │ 6   │ API spec update               │ docs/API_SPEC.md                                                             │ Small  │
  └─────┴───────────────────────────────┴──────────────────────────────────────────────────────────────────────────────┴────────┘

  Total: ~3-4 focused sessions of work.

Good — get_by_api_key already filters is_active == True at the DB level, and get_active_api_key adds the expires_at check on top.
All claims accurate.

Verification Report:

┌──────────────────────────────────────────────────────────────────┬─────────────────────────────────────────────┬──────────┐
│ Claim │ Source │ Verified │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ make init-agent target exists with ADDRESS param │ Makefile │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ make init-agent accepts optional NAME param │ Makefile │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ scripts/create_agent.py accepts positional address + --name flag │ argparse in script │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ POST /v1/auth/token route registered │ token.py:30, router.py:10 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Access token TTL = 1 hour │ AGENT_ACCESS_TOKEN_TTL = timedelta(hours=1) │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Refresh token TTL = 30 days │ REFRESH_TOKEN_EXPIRY = timedelta(days=30) │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ expires_in = 3600 in response │ int(AGENT_ACCESS_TOKEN_TTL.total_seconds()) │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ token_source: "api_key" in JWT claims │ token.py:59,65 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ verified_factors: [] in JWT claims │ token.py:60,66 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Inactive keys rejected (401) │ get_by_api_key filters is_active == True │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Expired keys rejected │ get_active_api_key checks expires_at │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ ttl: Optional[timedelta] param on create_access_token │ token_manager.py:87 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Default TTL fallback to ACCESS_TOKEN_EXPIRY (30 days) │ token_manager.py:106 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Spec file change: get_active_api_key method name │ api_key.py:102 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ Spec file change: ttl: Optional[timedelta] param name │ token_manager.py:87 │ Yes │
├──────────────────────────────────────────────────────────────────┼─────────────────────────────────────────────┼──────────┤
│ API_SPEC refresh example (no Auth header needed) │ refresh.py:27-29 — no auth dependency │ Yes │
└──────────────────────────────────────────────────────────────────┴─────────────────────────────────────────────┴──────────┘

Claims verified: 16
False statements found: 0
Uncertain items: 0

All output is accurate.

Agent Compatibility Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Enable autonomous AI agents to authenticate and use all dashboard-api endpoints without human-interactive flows.

Architecture: New POST /v1/auth/token endpoint exchanges a gw_ API key for a 1-hour JWT. A headless bootstrap script creates agent users + API keys directly in the DB. No changes to existing route handlers — once agents have a JWT, they use the same endpoints as dashboard users.

Tech Stack: FastAPI, SQLAlchemy, authlib (JWT), pytest, httpx

Spec: docs/superpowers/specs/2026-03-18-agent-compatibility-design.md

---

Task 1: Add ttl_seconds parameter to TokenManager.create_access_token

Files:

• Modify: app/services/tokens/token_manager.py:83-113

• Test: tests/unit/services/tokens/test_token_manager.py

• Step 1: Write failing test for custom TTL

In tests/unit/services/tokens/test_token_manager.py, add:

@pytest.mark.asyncio(loop_scope="session")
async def test_create_access_token_custom_ttl(redis, token_manager):
    """Access token with custom TTL uses the provided expiry instead of default."""
    token_str = await token_manager.create_access_token(
        data={"sub": "test-user-id"},
        ttl=timedelta(hours=1),
    )
    decoded = await token_manager.decode_token(token_str)
    # Token should expire within ~1 hour, not ~30 days
    exp_timestamp = decoded.exp
    now = datetime.now(UTC).timestamp()
    ttl_seconds = exp_timestamp - now
    assert ttl_seconds < 3700  # ~1 hour with some tolerance
    assert ttl_seconds > 3500

• Step 2: Run test to verify it fails

Run: pytest tests/unit/services/tokens/test_token_manager.py::test_create_access_token_custom_ttl -v
Expected: FAIL — create_access_token() does not accept ttl parameter.

• Step 3: Add ttl parameter to create_access_token

In app/services/tokens/token_manager.py, modify create_access_token (line 83):

async def create_access_token(
    self,
    data: Dict[str, str],
    verified_factors: Optional[List[Dict[str, str]]] = None,
    ttl: Optional[datetime.timedelta] = None,
) -> str:
    """Creates an access token with embedded verified factors.

    Args:
        data: The claims to include in the access token.
        verified_factors: A list of factor JWTs.
        ttl: Custom token lifetime. Defaults to ACCESS_TOKEN_EXPIRY (30 days).

    Returns:
        The generated JWT access token.
    """
    expiry = ttl if ttl is not None else ACCESS_TOKEN_EXPIRY
    factors_list = (
        verified_factors if verified_factors is not None else data.pop("verified_factors", [])
    )
    to_encode = Token(
        token_type=TokenType.ACCESS_TOKEN,
        exp=datetime.datetime.now(datetime.UTC) + expiry,
        iat=datetime.datetime.now(datetime.UTC),
        verified_factors=[FactorInfo(**f) for f in factors_list],
        **self.base_claims,
        **data,
    )

    return jwt.encode(
        {"kid": self.kid, "alg": "RS256"}, to_encode.model_dump(), self.private_key
    ).decode("utf-8")

• Step 4: Run test to verify it passes

Run: pytest tests/unit/services/tokens/test_token_manager.py::test_create_access_token_custom_ttl -v
Expected: PASS

• Step 5: Verify existing tests still pass

Run: pytest tests/unit/services/tokens/test_token_manager.py -v
Expected: All existing tests PASS (no behavioral change for callers that don't pass ttl).

• Step 6: Commit

git add app/services/tokens/token_manager.py tests/unit/services/tokens/test_token_manager.py
git commit -m "feat: add ttl parameter to TokenManager.create_access_token"

---

Task 2: Add get_by_api_key expiry check to ApiKeyService

The get_by_api_key method already exists at app/services/api_key.py:89-100. It checks is_active but does not check expires_at. The token exchange endpoint needs expiry validation.

Files:

• Modify: app/services/api_key.py:89-100

• Test: tests/unit/services/test_api_key_service.py (new file)

• Step 1: Write failing test for expired key rejection

Create tests/unit/services/__init__.py if it doesn't exist, then create tests/unit/services/test_api_key_service.py:

import datetime
from unittest.mock import AsyncMock
from uuid import UUID

import pytest

from app.db.models.api_key import ApiKey
from app.services.api_key import ApiKeyService


def _make_key(**overrides):
    defaults = dict(
        id=UUID(int=1),
        user_id=UUID(int=2),
        address="0x" + "ab" * 20,
        api_key="gw_test",
        name="test",
        permissions=["rpc_write"],
        is_active=True,
        expires_at=None,
        created_at=datetime.datetime(2019, 1, 1, tzinfo=datetime.UTC),
        updated_at=datetime.datetime(2019, 1, 1, tzinfo=datetime.UTC),
    )
    defaults.update(overrides)
    return ApiKey(**defaults)


@pytest.mark.asyncio
async def test_get_active_api_key_excludes_expired(mocker):
    """get_active_api_key returns None for expired keys."""
    expired_key = _make_key(
        api_key="gw_expired",
        expires_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC),
    )
    mocker.patch.object(
        ApiKeyService, "get_by_api_key", new_callable=AsyncMock, return_value=expired_key
    )

    service = ApiKeyService()
    result = await service.get_active_api_key("gw_expired")
    assert result is None


@pytest.mark.asyncio
async def test_get_active_api_key_returns_valid_key(mocker):
    """get_active_api_key returns a valid, non-expired key."""
    valid_key = _make_key(api_key="gw_valid")
    mocker.patch.object(
        ApiKeyService, "get_by_api_key", new_callable=AsyncMock, return_value=valid_key
    )

    service = ApiKeyService()
    result = await service.get_active_api_key("gw_valid")
    assert result is not None
    assert result.api_key == "gw_valid"


@pytest.mark.asyncio
async def test_get_active_api_key_returns_none_for_missing_key(mocker):
    """get_active_api_key returns None when key does not exist."""
    mocker.patch.object(
        ApiKeyService, "get_by_api_key", new_callable=AsyncMock, return_value=None
    )

    service = ApiKeyService()
    result = await service.get_active_api_key("gw_nonexistent")
    assert result is None

• Step 2: Run tests to verify they fail

Run: pytest tests/unit/services/test_api_key_service.py -v
Expected: FAIL — get_active_api_key method does not exist.

• Step 3: Add get_active_api_key method

In app/services/api_key.py, add after the existing get_by_api_key method (after line 100):

async def get_active_api_key(self, api_key_value: str) -> Optional[ApiKey]:
    """Get an active, non-expired API key by its value.

    Unlike get_by_api_key, this also checks expires_at.
    Used for agent token exchange where expiry must be enforced.
    """
    api_key = await self.get_by_api_key(api_key_value)
    if not api_key:
        return None
    if api_key.expires_at and api_key.expires_at < datetime.datetime.now(datetime.UTC):
        return None
    return api_key

• Step 4: Run tests to verify they pass

Run: pytest tests/unit/services/test_api_key_service.py -v
Expected: PASS

• Step 5: Commit

git add app/services/api_key.py tests/unit/services/test_api_key_service.py
git commit -m "feat: add get_active_api_key with expiry check to ApiKeyService"

---

Task 3: Create token exchange endpoint POST /v1/auth/token

Files:

• Create: app/api/v1/routes/authn/token.py

• Modify: app/api/v1/routes/authn/router.py:1-11 (register new route)

• Test: tests/unit/api/v1/routes/authn/test_token.py

• Step 1: Write failing tests for the token exchange endpoint

Create tests/unit/api/v1/routes/authn/test_token.py:

import datetime
import secrets
from unittest.mock import AsyncMock
from uuid import UUID

import pytest
from fastapi import status

from app.db.models.api_key import ApiKey

USER_ID = UUID(int=1)
WALLET = f"0x{secrets.token_hex(20)}"
_NOW = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)


def _make_api_key(**overrides):
    defaults = dict(
        id=UUID(int=10),
        user_id=USER_ID,
        address=WALLET,
        api_key="gw_test123",
        name="agent-key",
        permissions=["admin"],
        rate_limit=None,
        is_active=True,
        expires_at=None,
        description=None,
        created_at=_NOW,
        updated_at=_NOW,
    )
    defaults.update(overrides)
    return ApiKey(**defaults)


@pytest.fixture
def mock_get_active_api_key(mocker):
    return mocker.patch(
        "app.services.api_key.ApiKeyService.get_active_api_key",
        new_callable=AsyncMock,
        return_value=_make_api_key(),
    )


class TestTokenExchange:
    def test_exchange_valid_api_key_returns_tokens(
        self, client, redis, mock_get_active_api_key
    ):
        """Valid gw_ API key returns access_token and refresh_token."""
        response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": "gw_test123"},
        )
        assert response.status_code == status.HTTP_200_OK
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data
        assert data["expires_in"] == 3600
        assert data["token_type"] == "Bearer"

    def test_exchange_missing_api_key_returns_401(self, client, redis):
        """Missing X-API-Key header returns 401."""
        response = client.post("/v1/auth/token")
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

    def test_exchange_invalid_api_key_returns_401(self, client, redis, mocker):
        """Invalid API key returns 401."""
        mocker.patch(
            "app.services.api_key.ApiKeyService.get_active_api_key",
            return_value=None,
        )
        response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": "gw_invalid"},
        )
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

    def test_exchange_expired_api_key_returns_401(self, client, redis, mocker):
        """Expired API key returns 401."""
        mocker.patch(
            "app.services.api_key.ApiKeyService.get_active_api_key",
            return_value=None,
        )
        response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": "gw_expired"},
        )
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

    @pytest.mark.asyncio
    async def test_exchange_token_has_api_key_source(
        self, client, redis, mock_get_active_api_key
    ):
        """Exchanged token includes token_source=api_key."""
        from app.services.tokens.token_manager import TokenManager

        response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": "gw_test123"},
        )
        assert response.status_code == status.HTTP_200_OK
        data = response.json()

        # Decode the token and verify claims
        tm = TokenManager()
        decoded = await tm.decode_token(data["access_token"])
        assert decoded.token_source == "api_key"
        assert str(decoded.sub) == str(USER_ID)
        assert decoded.verified_factors == []

• Step 2: Run tests to verify they fail

Run: pytest tests/unit/api/v1/routes/authn/test_token.py -v
Expected: FAIL — module does not exist, route not registered.

• Step 3: Create the token exchange route

Create app/api/v1/routes/authn/token.py:

import datetime
import logging
from typing import Optional

from fastapi import APIRouter
from fastapi import Depends
from fastapi import Header
from fastapi import HTTPException
from fastapi import status
from pydantic import BaseModel

from app.services.api_key import ApiKeyService
from app.services.tokens.token_manager import TokenManager

logger = logging.getLogger(__name__)

AGENT_ACCESS_TOKEN_TTL = datetime.timedelta(hours=1)

router = APIRouter(tags=["authn"])


class TokenExchangeResponse(BaseModel):
    access_token: str
    refresh_token: str
    expires_in: int
    token_type: str = "Bearer"


@router.post(
    "/v1/auth/token",
    response_model=TokenExchangeResponse,
    status_code=status.HTTP_200_OK,
)
async def exchange_api_key_for_token(
    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
    api_key_service: ApiKeyService = Depends(ApiKeyService),
    token_manager: TokenManager = Depends(TokenManager),
) -> TokenExchangeResponse:
    """Exchange a gw_ API key for a short-lived JWT access token.

    Agents use this endpoint to obtain JWTs for authenticating against
    all other dashboard endpoints. The returned access token expires
    in 1 hour; use POST /v1/auth/refresh to rotate.
    """
    if not x_api_key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="X-API-Key header is required",
        )

    api_key = await api_key_service.get_active_api_key(x_api_key)
    if not api_key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid or expired API key",
        )

    access_token = await token_manager.create_access_token(
        data={"sub": str(api_key.user_id), "token_source": "api_key"},
        verified_factors=[],
        ttl=AGENT_ACCESS_TOKEN_TTL,
    )

    refresh_token = await token_manager.create_refresh_token(
        data={"sub": str(api_key.user_id), "token_source": "api_key"},
        verified_factors=[],
    )

    return TokenExchangeResponse(
        access_token=access_token,
        refresh_token=refresh_token,
        expires_in=int(AGENT_ACCESS_TOKEN_TTL.total_seconds()),
    )

• Step 4: Register the route in authn router

In app/api/v1/routes/authn/router.py, add the import and include:

from fastapi import APIRouter

from .cli_token import router as cli_token_router
from .factor.router import router as factor_router
from .session.router import router as session_router
from .token import router as token_router

router = APIRouter()

router.include_router(token_router)
router.include_router(cli_token_router)
router.include_router(factor_router)
router.include_router(session_router)

• Step 5: Run tests to verify they pass

Run: pytest tests/unit/api/v1/routes/authn/test_token.py -v
Expected: All PASS

• Step 6: Run full test suite to check for regressions

Run: pytest tests/unit/ -v
Expected: All PASS

• Step 7: Commit

git add app/api/v1/routes/authn/token.py app/api/v1/routes/authn/router.py tests/unit/api/v1/routes/authn/test_token.py
git commit -m "feat: add POST /v1/auth/token endpoint for API key to JWT exchange"

---

Task 4: Create headless bootstrap script

Files:

• Create: scripts/create_agent.py

• Modify: Makefile:51-61 (add init-agent target)

• Test: Manual verification (script operates directly on DB)

• Step 1: Create the bootstrap script

Create scripts/create_agent.py:

"""
Headless bootstrap script for creating agent users with API keys.

Creates a User record with a wallet address and an API key directly
in the database, bypassing interactive authentication flows.

Usage:
    python scripts/create_agent.py 0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18
    python scripts/create_agent.py 0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18 --name my-agent
"""

import argparse
import asyncio
import logging
import re
import sys

from sqlalchemy import select

from app.db.models.api_key import ApiKey
from app.db.models.user import User
from app.db.session import async_session
from app.utils.api_key import generate_api_key

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)

ADDRESS_RE = re.compile(r"^0x[a-fA-F0-9]{40}$")


async def create_agent(address: str, name: str) -> None:
    """Create an agent user with an API key."""
    if not ADDRESS_RE.match(address):
        logger.error("invalid Ethereum address: %s", address)
        sys.exit(1)

    async with async_session() as session:
        # Check if a user with this address already exists
        result = await session.execute(
            select(User).where(User.public_address == address)
        )
        user = result.scalar_one_or_none()

        if user:
            logger.info("user already exists for address %s (id: %s)", address, user.id)
        else:
            user = User(public_address=address, is_active=True)
            session.add(user)
            await session.flush()
            logger.info("created user %s for address %s", user.id, address)

        # Check for existing active API key
        result = await session.execute(
            select(ApiKey).where(
                ApiKey.user_id == user.id,
                ApiKey.is_active == True,
            )
        )
        existing_key = result.scalars().first()

        if existing_key:
            logger.info("active API key already exists: %s", existing_key.api_key)
            api_key_value = existing_key.api_key
        else:
            api_key_value = generate_api_key()
            api_key = ApiKey(
                user_id=user.id,
                address=address,
                api_key=api_key_value,
                name=name,
                permissions=["admin"],
                is_active=True,
            )
            session.add(api_key)
            logger.info("created API key: %s", api_key_value)

        await session.commit()

    print()
    print("=" * 60)
    print("Agent Setup Complete")
    print("=" * 60)
    print(f"  Address:  {address}")
    print(f"  User ID:  {user.id}")
    print(f"  API Key:  {api_key_value}")
    print(f"  Name:     {name}")
    print()
    print("Exchange for JWT:")
    print(f"  curl -X POST <BASE_URL>/v1/auth/token \\")
    print(f"    -H 'X-API-Key: {api_key_value}'")
    print("=" * 60)


def main():
    parser = argparse.ArgumentParser(description="Create an agent user with API key")
    parser.add_argument("address", help="Ethereum address (0x-prefixed, 40 hex chars)")
    parser.add_argument("--name", default="agent", help="API key name (default: agent)")
    args = parser.parse_args()

    asyncio.run(create_agent(args.address, args.name))


if __name__ == "__main__":
    main()

• Step 2: Add Makefile target

In Makefile, add before the .PHONY line (after the init target):

# Target to create an agent user with API key (headless, no interactive auth)
# Usage: make init-agent ADDRESS=0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18
# Optional: make init-agent ADDRESS=0x... NAME=my-agent
init-agent:
	@if [ -z "$(ADDRESS)" ]; then \
		echo "Error: ADDRESS is required"; \
		echo "Usage: make init-agent ADDRESS=0x..."; \
		exit 1; \
	fi
	docker compose exec runtime python scripts/create_agent.py $(ADDRESS) $(if $(NAME),--name $(NAME),)

Update .PHONY to include init-agent:

.PHONY: venv activate install dev runtime test down clean init init-agent

• Step 3: Commit

git add scripts/create_agent.py Makefile
git commit -m "feat: add make init-agent for headless agent bootstrapping"

---

Task 5: Update API spec documentation

Files:

• Modify: docs/API_SPEC.md

• Step 1: Add Agent Authentication section after the DPoP section

In docs/API_SPEC.md, after the DPoP section (around line 147), add:

### API Key Token Exchange (Agent Authentication)

Headless agents authenticate by exchanging a `gw_` API key for a short-lived JWT:

1. Obtain a `gw_` API key (via dashboard UI or `make init-agent`)
2. Call `POST /v1/auth/token` with the key in the `X-API-Key` header
3. Use the returned `access_token` as a Bearer token for all authenticated endpoints
4. When the access token expires (1 hour), call `POST /v1/auth/refresh` with the `refresh_token`

Agent Server
| |
| POST /v1/auth/token |
| X-API-Key: gw_abc123... |
|----- ------------------------------>|
| |
| { access_token, refresh_token, |
| expires_in: 3600 } |
|<------------------------------------|
| |
| GET /v1/policy-client-owner |
| Authorization: Bearer |
|------------------------------------>|
| (normal JWT auth flow) |
|<------------------------------------|

Agent tokens have `token_source: "api_key"` and empty `verified_factors`. They grant
full access to all endpoints the API key's owning user can access.

• Step 2: Add the Token Exchange endpoint section

After the CLI Token section and before User Management, add the endpoint documentation:

### Token Exchange

#### POST /v1/auth/token

Exchange a `gw_` API key for a short-lived JWT access token and refresh token. Designed for headless agent authentication.

**Authentication:** API key via `X-API-Key` header (no JWT required)

**Headers:**

| Header     | Type   | Required | Description                   |
|------------|--------|----------|-------------------------------|
| `X-API-Key`| string | Yes      | A valid, active `gw_` API key |

**Response:** `200 OK`

```json
{
  "access_token": "eyJhbGciOiJSUzI1NiIs...",
  "refresh_token": "eyJhbGciOiJSUzI1NiIs...",
  "expires_in": 3600,
  "token_type": "Bearer"
}

Field	Type	Description
access_token	string	JWT access token (1-hour TTL)
refresh_token	string	JWT refresh token (30-day TTL)
expires_in	integer	Access token lifetime in seconds (3600)
token_type	string	Always "Bearer"

Errors:

• 401 - Missing, invalid, inactive, or expired API key

- [ ] **Step 3: Add agent bootstrap instructions to Integration Guide**

At the end of the Integration Guide section, add:

```markdown
### Agent Integration (Headless)

For autonomous agents that cannot complete interactive authentication:

1. **Bootstrap:** An admin creates the agent user and API key:
   ```bash
   make init-agent ADDRESS=0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18 NAME=my-agent

2. Authenticate: The agent exchanges its API key for a JWT:

   curl -X POST https://dashboard.api.newton.xyz/v1/auth/token \
     -H "X-API-Key: gw_YOUR_KEY_HERE"

3. Use: All subsequent calls use the JWT as a Bearer token, identical to dashboard UI flow.

4. Refresh: When the access token expires (1 hour), refresh it:

   curl -X POST https://dashboard.api.newton.xyz/v1/auth/refresh \
     -H "Content-Type: application/json" \
     -d '{"refresh_token": "eyJ..."}'

- [ ] **Step 4: Commit**

```bash
git add docs/API_SPEC.md
git commit -m "docs: add agent authentication flow and POST /v1/auth/token to API spec"

---

Task 6: Integration test — full agent auth flow

Files:

• Create: tests/integration/api/v1/routes/authn/test_token.py

• Step 1: Write integration test for the full exchange flow

Create tests/integration/api/v1/routes/authn/__init__.py (if it doesn't exist) and tests/integration/api/v1/routes/authn/test_token.py:

import datetime
import secrets
from uuid import uuid4

import pytest
from fastapi import status

from app.db.models.api_key import ApiKey
from app.db.models.user import User
from app.db.session import async_session
from app.services.tokens.token_manager import TokenManager
from app.utils.api_key import generate_api_key


WALLET = f"0x{secrets.token_hex(20)}"


@pytest.fixture
async def agent_user():
    """Create a user with a wallet address and an API key."""
    async with async_session() as session:
        user = User(public_address=WALLET, is_active=True)
        session.add(user)
        await session.flush()

        api_key_value = generate_api_key()
        api_key = ApiKey(
            user_id=user.id,
            address=WALLET,
            api_key=api_key_value,
            name="test-agent",
            permissions=["admin"],
            is_active=True,
        )
        session.add(api_key)
        await session.commit()
        await session.refresh(user)
        await session.refresh(api_key)

    return {"user": user, "api_key_value": api_key_value}


class TestTokenExchangeIntegration:
    @pytest.mark.asyncio
    async def test_full_agent_auth_flow(self, client, agent_user):
        """Agent can exchange API key for JWT and use it to call a protected endpoint."""
        api_key_value = agent_user["api_key_value"]
        user = agent_user["user"]

        # Step 1: Exchange API key for JWT
        exchange_response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": api_key_value},
        )
        assert exchange_response.status_code == status.HTTP_200_OK
        tokens = exchange_response.json()
        access_token = tokens["access_token"]
        assert tokens["expires_in"] == 3600

        # Step 2: Use the JWT to call a protected endpoint
        user_response = client.get(
            "/v1/user",
            headers={"Authorization": f"Bearer {access_token}"},
        )
        assert user_response.status_code == status.HTTP_200_OK

    @pytest.mark.asyncio
    async def test_exchange_then_refresh(self, client, agent_user):
        """Agent can exchange API key for JWT and then refresh the tokens."""
        api_key_value = agent_user["api_key_value"]

        # Exchange
        exchange_response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": api_key_value},
        )
        tokens = exchange_response.json()

        # Refresh
        refresh_response = client.post(
            "/v1/auth/refresh",
            json={"refresh_token": tokens["refresh_token"]},
        )
        assert refresh_response.status_code == status.HTTP_200_OK
        new_tokens = refresh_response.json()
        assert "access_token" in new_tokens

    @pytest.mark.asyncio
    async def test_deactivated_key_cannot_exchange(self, client, agent_user):
        """Deactivated API key is rejected at exchange."""
        api_key_value = agent_user["api_key_value"]

        # Deactivate the key
        async with async_session() as session:
            from sqlalchemy import update
            await session.execute(
                update(ApiKey)
                .where(ApiKey.api_key == api_key_value)
                .values(is_active=False)
            )
            await session.commit()

        response = client.post(
            "/v1/auth/token",
            headers={"X-API-Key": api_key_value},
        )
        assert response.status_code == status.HTTP_401_UNAUTHORIZED

• Step 2: Run integration tests

Run: docker compose run --rm test sh -c "alembic upgrade head && pytest tests/integration/api/v1/routes/authn/test_token.py -v"
Expected: All PASS

• Step 3: Commit

git add tests/integration/api/v1/routes/authn/
git commit -m "test: add integration tests for agent token exchange flow"

---

Task 7: Run full test suite and verify coverage

• Step 1: Run the complete test suite in Docker

Run: make test
Expected: All tests pass, coverage >= 70%.

• Step 2: Fix any failures if they arise

If any tests fail, fix the root cause (import errors, mock mismatches, etc.) and re-run.

• Step 3: Final commit with any fixes

Only commit if fixes were needed:

git add -u
git commit -m "fix: address test failures from agent compatibility changes"