| 1 | from __future__ import annotations |
| 2 | |
| 3 | from typing import Annotated |
| 4 | |
| 5 | from fastapi import Depends, Header, HTTPException |
| 6 | |
| 7 | from open_agent_registry.auth import hash_secret |
| 8 | from open_agent_registry.db import Database, row_to_agent |
| 9 | |
| 10 | |
| 11 | def get_db() -> Database: |
| 12 | from open_agent_registry.app import db |
| 13 | |
| 14 | return db |
| 15 | |
| 16 | |
| 17 | def require_agent( |
| 18 | authorization: Annotated[str | None, Header()] = None, |
| 19 | db: Database = Depends(get_db), |
| 20 | ) -> dict: |
| 21 | if not authorization or not authorization.startswith("Bearer "): |
| 22 | raise HTTPException(status_code=401, detail="Missing Bearer API key") |
| 23 | api_key = authorization.removeprefix("Bearer ").strip() |
| 24 | if not api_key: |
| 25 | raise HTTPException(status_code=401, detail="Empty API key") |
| 26 | key_hash = hash_secret(api_key) |
| 27 | with db.connect() as conn: |
| 28 | row = conn.execute( |
| 29 | "SELECT * FROM agents WHERE api_key_hash = ?", |
| 30 | (key_hash,), |
| 31 | ).fetchone() |
| 32 | if row is None: |
| 33 | raise HTTPException(status_code=401, detail="Invalid API key") |
| 34 | return row_to_agent(row) |
| 35 | |
View only · write via MCP/CIDE