diff --git a/src/fabledassistant/services/api_keys.py b/src/fabledassistant/services/api_keys.py
new file mode 100644
index 0000000..5560838
--- /dev/null
+++ b/src/fabledassistant/services/api_keys.py
@@ -0,0 +1,87 @@
+import hashlib
+import secrets
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+
+from fabledassistant.models import async_session
+from fabledassistant.models.api_key import ApiKey
+
+
+def generate_key() -> str:
+    """Generate a new full API key. Never stored — caller must hash it."""
+    return "fmcp_" + secrets.token_urlsafe(32)
+
+
+def _hash_key(key: str) -> str:
+    return hashlib.sha256(key.encode()).hexdigest()
+
+
+def _key_prefix(key: str) -> str:
+    """Return first 12 chars of the key for display (e.g. 'fmcp_abcdefg')."""
+    return key[:12]
+
+
+async def create_api_key(
+    user_id: int, name: str, scope: str
+) -> tuple[str, dict]:
+    """Create a new API key. Returns (full_key, key_dict). full_key is never stored."""
+    if scope not in ("read", "write"):
+        raise ValueError("scope must be 'read' or 'write'")
+    full_key = generate_key()
+    key = ApiKey(
+        user_id=user_id,
+        name=name,
+        key_hash=_hash_key(full_key),
+        key_prefix=_key_prefix(full_key),
+        scope=scope,
+    )
+    async with async_session() as session:
+        session.add(key)
+        await session.commit()
+        await session.refresh(key)
+        return full_key, key.to_dict()
+
+
+async def list_api_keys(user_id: int) -> list[dict]:
+    """List all non-revoked API keys for the user."""
+    async with async_session() as session:
+        result = await session.execute(
+            select(ApiKey)
+            .where(ApiKey.user_id == user_id, ApiKey.revoked_at.is_(None))
+            .order_by(ApiKey.created_at.desc())
+        )
+        return [k.to_dict() for k in result.scalars().all()]
+
+
+async def revoke_api_key(user_id: int, key_id: int) -> bool:
+    """Soft-delete a key by setting revoked_at. Returns True if found."""
+    async with async_session() as session:
+        result = await session.execute(
+            select(ApiKey).where(ApiKey.id == key_id, ApiKey.user_id == user_id)
+        )
+        key = result.scalars().first()
+        if key is None:
+            return False
+        key.revoked_at = datetime.now(timezone.utc)
+        await session.commit()
+        return True
+
+
+async def lookup_key(raw_key: str) -> ApiKey | None:
+    """Look up a non-revoked ApiKey by raw token value. Updates last_used_at."""
+    key_hash = _hash_key(raw_key)
+    async with async_session() as session:
+        result = await session.execute(
+            select(ApiKey).where(
+                ApiKey.key_hash == key_hash,
+                ApiKey.revoked_at.is_(None),
+            )
+        )
+        key = result.scalars().first()
+        if key is None:
+            return None
+        key.last_used_at = datetime.now(timezone.utc)
+        await session.commit()
+        await session.refresh(key)
+        return key
diff --git a/tests/test_api_keys.py b/tests/test_api_keys.py
new file mode 100644
index 0000000..c9f4842
--- /dev/null
+++ b/tests/test_api_keys.py
@@ -0,0 +1,166 @@
+"""Unit tests for api_keys service — pure logic, DB calls mocked."""
+import hashlib
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from fabledassistant.services.api_keys import (
+    _hash_key,
+    _key_prefix,
+    generate_key,
+    create_api_key,
+    list_api_keys,
+    revoke_api_key,
+    lookup_key,
+)
+
+
+def test_generate_key_format():
+    key = generate_key()
+    assert key.startswith("fmcp_")
+    assert len(key) > 12
+
+
+def test_generate_key_uniqueness():
+    keys = {generate_key() for _ in range(100)}
+    assert len(keys) == 100
+
+
+def test_hash_key_is_sha256():
+    key = "fmcp_testkey"
+    h = _hash_key(key)
+    expected = hashlib.sha256(key.encode()).hexdigest()
+    assert h == expected
+
+
+def test_key_prefix_returns_first_12_chars():
+    key = "fmcp_abcdefghijklmnop"
+    assert _key_prefix(key) == "fmcp_abcdefg"
+
+
+@pytest.mark.asyncio
+async def test_create_api_key_returns_full_key():
+    mock_key_obj = MagicMock()
+    mock_key_obj.id = 1
+    mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}
+
+    with patch("fabledassistant.services.api_keys.async_session") as mock_session_ctx:
+        mock_session = AsyncMock()
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=False)
+        mock_session.add = MagicMock()
+        mock_session.commit = AsyncMock()
+        mock_session.refresh = AsyncMock(side_effect=lambda obj: None)
+        # Return our fake object from refresh
+        mock_session.refresh = AsyncMock()
+        mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}
+        mock_session_ctx.return_value = mock_session
+
+        # Patch the select result
+        with patch("fabledassistant.services.api_keys.ApiKey") as mock_model:
+            mock_model.return_value = mock_key_obj
+            full_key, key_dict = await create_api_key(user_id=1, name="test", scope="read")
+
+    assert full_key.startswith("fmcp_")
+
+
+@pytest.mark.asyncio
+async def test_lookup_key_returns_none_for_unknown():
+    with patch("fabledassistant.services.api_keys.async_session") as mock_session_ctx:
+        mock_session = AsyncMock()
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=False)
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.first.return_value = None
+        mock_session.execute = AsyncMock(return_value=mock_result)
+        mock_session_ctx.return_value = mock_session
+
+        result = await lookup_key("fmcp_doesnotexist")
+
+    assert result is None
+
+
+def test_hash_key_deterministic():
+    key = "fmcp_some_test_key_value"
+    assert _hash_key(key) == _hash_key(key)
+
+
+@pytest.mark.asyncio
+async def test_scope_validation():
+    with pytest.raises(ValueError, match="scope must be"):
+        await create_api_key(1, "bad", "superadmin")
+
+
+# --- Auth middleware tests ---
+
+@pytest.mark.asyncio
+async def test_read_only_key_blocked_on_post():
+    """Read-only API key returns 403 on non-GET requests."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+    from quart import Quart
+
+    fake_key = MagicMock()
+    fake_key.user_id = 7
+    fake_key.scope = "read"
+
+    fake_user = MagicMock()
+    fake_user.role = "user"
+
+    app = Quart(__name__)
+
+    async def dummy():
+        from quart import jsonify
+        return jsonify({"ok": True})
+
+    from fabledassistant.auth import _check_auth
+    wrapped = _check_auth(dummy)
+
+    async with app.test_request_context(
+        "/test", method="POST",
+        headers={"Authorization": "Bearer fmcp_readonly"},
+    ):
+        with patch("fabledassistant.auth.lookup_key", AsyncMock(return_value=fake_key)), \
+             patch("fabledassistant.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
+            resp = await wrapped()
+    assert resp[1] == 403
+
+
+@pytest.mark.asyncio
+async def test_bearer_token_valid_write_key_passes():
+    """Valid write-scoped bearer token passes auth on POST."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+    from quart import Quart, g
+
+    fake_key = MagicMock()
+    fake_key.user_id = 7
+    fake_key.scope = "write"
+
+    fake_user = MagicMock()
+    fake_user.role = "user"
+    fake_user.id = 7
+
+    app = Quart(__name__)
+
+    result_holder = {}
+
+    async def dummy():
+        result_holder["user"] = g.user
+        result_holder["api_key"] = g.api_key
+        from quart import jsonify
+        return jsonify({"ok": True})
+
+    from fabledassistant.auth import _check_auth
+    wrapped = _check_auth(dummy)
+
+    async with app.test_request_context(
+        "/test", method="POST",
+        headers={"Authorization": "Bearer fmcp_writekey"},
+    ):
+        with patch("fabledassistant.auth.lookup_key", AsyncMock(return_value=fake_key)), \
+             patch("fabledassistant.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
+            resp = await wrapped()
+
+    # Should not be a tuple (no error response)
+    assert not isinstance(resp, tuple) or resp[1] == 200
+    assert result_holder.get("user") is fake_user
+    assert result_holder.get("api_key") is fake_key