FastAPI backend for managing an organizational catalog of AI-agent tools. Supports two tool types — function tools (defined by JSON Schema) and remote tools (external MCP servers verified via HTTP probing) — with JWT authentication, role-based access control, encrypted secret storage, comprehensive audit logging, and rate limiting.
- Architecture
- Tech Stack
- Project Structure
- Getting Started
- API Reference
- Authentication & Authorization
- Tool Types
- Remote Tool Probing
- Security
- Database
- Testing
- Rate Limiting
- Environment Variables
- License
The application follows a layered architecture with strict separation of concerns:
┌─────────────────────────────────────┐ │ API Layer (app/api/) │ Routers, request/response handling ├─────────────────────────────────────┤ │ Service Layer (app/services/) │ Business logic, orchestration ├─────────────────────────────────────┤ │ Repository Layer (app/repos/) │ Data access, SQL queries ├─────────────────────────────────────┤ │ Model Layer (app/models/) │ ORM models, table definitions ├─────────────────────────────────────┤ │ Database (SQLite) │ Persistent storage └─────────────────────────────────────┘ Key design principles:
- Each layer only communicates with its immediate neighbor.
- Dependency injection via FastAPI's
Depends()wires all layers together at request time. - Pydantic schemas validate all inbound data and shape all outbound responses.
- Transactions are managed at the service layer — repositories never commit.
| Component | Technology |
|---|---|
| Framework | FastAPI 0.135+ |
| ORM | SQLAlchemy 2.0+ (synchronous) |
| Migrations | Alembic 1.18+ |
| Validation | Pydantic 2.12+ / pydantic-settings |
| Auth | PyJWT (HS256 JWTs) |
| Password Hashing | argon2-cffi |
| Encryption | cryptography (Fernet symmetric) |
| HTTP Client | httpx (async, for remote probing) |
| Schema Validation | jsonschema (function tool input schemas) |
| Rate Limiting | limits (fixed-window strategy) |
| Database | SQLite (default, swappable via DATABASE_URL) |
| Python | 3.14+ |
├── app/ │ ├── main.py # FastAPI app factory, lifespan, middleware │ ├── cli.py # CLI for seeding admin users │ │ │ ├── core/ # Cross-cutting concerns │ │ ├── constants.py # Regex patterns, pagination defaults │ │ ├── enums.py # UserRole, ToolType, RiskLevel, ProbeStatus, ... │ │ ├── exceptions.py # Custom HTTP exceptions (401, 403, 404, 409, 422) │ │ ├── logging.py # Logging configuration │ │ ├── rate_limit.py # Rate limiter (fixed-window, in-memory) │ │ ├── security.py # PasswordManager, TokenManager, SecretManager │ │ ├── settings.py # Pydantic settings (env-driven configuration) │ │ └── time.py # UTC datetime helpers │ │ │ ├── db/ # Database infrastructure │ │ ├── base.py # SQLAlchemy Base, TimestampMixin, UUIDPrimaryKeyMixin │ │ ├── bootstrap.py # Auto-create schema on first startup │ │ └── session.py # Engine and session factory │ │ │ ├── models/ # SQLAlchemy ORM models (1 model = 1 table) │ │ ├── user.py # User (username, password_hash, role) │ │ ├── refresh_token.py # RefreshToken (token_id, expiry, revocation) │ │ ├── tool.py # Tool (name, type, risk_level, active) │ │ ├── function_tool_config.py # FunctionToolConfig (JSON input_schema) │ │ ├── remote_tool_config.py # RemoteToolConfig (url, transport, probe state) │ │ ├── remote_tool_connection_test.py # ConnectionTest (probe history) │ │ └── audit_event.py # AuditEvent (action, actor, outcome, metadata) │ │ │ ├── schemas/ # Pydantic request/response models │ │ ├── common.py # APIModel, PaginatedResponse, TimestampedResponse │ │ ├── auth.py # Login, Refresh, Elevate, CurrentUser │ │ ├── tool.py # ToolCreate, ToolUpdate, ToolResponse, ProbeResponse │ │ └── audit.py # AuditEventResponse │ │ │ ├── repositories/ # Data access layer (SQL queries) │ │ ├── user_repository.py # User CRUD │ │ ├── token_repository.py # Refresh token CRUD │ │ ├── tool_repository.py # Tool CRUD, filtering, pagination │ │ └── audit_repository.py # Audit event creation and listing │ │ │ ├── services/ # Business logic layer │ │ ├── auth_service.py # Login, refresh, elevate, bootstrap │ │ ├── tool_service.py # Tool CRUD, probe orchestration │ │ ├── probe_service.py # HTTP health checks for remote tools │ │ ├── audit_service.py # Audit event logging and querying │ │ └── resolver_service.py # Resolve active tools (future use) │ │ │ ├── api/ # FastAPI routers and DI wiring │ │ ├── dependencies.py # All Depends() factories, auth guards, rate limiters │ │ ├── serializers.py # Tool model → ToolResponse conversion │ │ ├── auth.py # /api/v1/auth/* routes │ │ ├── tools.py # /api/v1/tools/* routes │ │ └── audit.py # /api/v1/audit-events routes │ │ │ └── resolver/ # Resolver module (future integration) │ ├── alembic/ # Database migration scripts │ ├── env.py │ ├── script.py.mako │ └── versions/ │ └── 20260310_000001_initial_schema.py │ ├── tests/ │ ├── conftest.py # Fixtures (test client, admin user, mock transport) │ ├── unit/ │ │ └── test_schemas_and_security.py │ └── integration/ │ ├── test_auth_api.py │ └── test_tools_api.py │ ├── scripts/ │ └── e2e_smoke_test.sh # End-to-end smoke test (curl-based) │ ├── pyproject.toml # Project metadata, dependencies, tool config ├── alembic.ini # Alembic migration configuration └── AGENTS.md # AI agent coding instructions - Python 3.14+
- pip (or any PEP 517-compatible installer)
# Clone the repository git clone https://github.com/<your-username>/tool-registry-backend.git cd tool-registry-backend # Create and activate a virtual environment python -m venv .venv source .venv/bin/activate # macOS / Linux # .venv\Scripts\activate # Windows # Install dependencies pip install -e . # Install dev dependencies (testing, linting, type checking) pip install -e ".[dev]"All configuration is driven by environment variables (or a .env file in the project root). Create a .env file for local development:
# .env JWT_SECRET_KEY=your-super-secret-key-change-in-production ENCRYPTION_SECRET=another-secret-for-encrypting-tool-headers BOOTSTRAP_ADMIN_USERNAME=admin BOOTSTRAP_ADMIN_PASSWORD=AdminPass123!See Environment Variables for the full list.
# Development mode with hot-reload uvicorn app.main:app --reload --port 8000The API is now available at http://127.0.0.1:8000. Interactive docs live at:
- Swagger UI:
http://127.0.0.1:8000/docs - ReDoc:
http://127.0.0.1:8000/redoc
An admin user can be created in two ways:
-
Automatic (via environment): Set
BOOTSTRAP_ADMIN_USERNAMEandBOOTSTRAP_ADMIN_PASSWORD— the user is created on first startup if they don't already exist. -
Manual (via CLI):
python -m app.cli seed-admin --username admin --password "AdminPass123!"
| Method | Endpoint | Description | Auth Required |
|---|---|---|---|
POST | /api/v1/auth/login | Authenticate and receive token pair | No |
POST | /api/v1/auth/refresh | Exchange refresh token for new token pair | No |
GET | /api/v1/auth/me | Get current user info | Bearer token |
POST | /api/v1/auth/elevate | Get short-lived elevated admin token | Bearer token (admin) |
Login request:
{ "username": "admin", "password": "AdminPass123!" }Login response:
{ "token_type": "bearer", "access_token": "eyJhbGciOiJIUzI1NiIs...", "access_token_expires_at": "2026-03-12T15:30:00Z", "refresh_token": "eyJhbGciOiJIUzI1NiIs...", "refresh_token_expires_at": "2026-03-26T15:15:00Z" }| Method | Endpoint | Description | Auth Required |
|---|---|---|---|
POST | /api/v1/tools | Create a tool (function or remote) | Admin |
GET | /api/v1/tools | List tools (with filters and pagination) | Any user |
GET | /api/v1/tools/{id} | Get a single tool by ID | Any user |
PATCH | /api/v1/tools/{id} | Update a tool | Admin |
POST | /api/v1/tools/{id}/activate | Activate a tool | Admin |
POST | /api/v1/tools/{id}/deactivate | Deactivate a tool | Admin |
DELETE | /api/v1/tools/{id} | Permanently delete a tool | Elevated admin |
POST | /api/v1/tools/remote/test-connection | Test remote tool connectivity | Admin |
POST | /api/v1/tools/{id}/reprobe | Re-probe a remote tool | Admin |
Create function tool:
{ "tool_type": "function", "name": "get_weather", "display_name": "Get Weather", "description": "Returns the current weather for a given city", "risk_level": "low", "input_schema": { "type": "object", "properties": { "city": { "type": "string" } }, "required": ["city"] } }Create remote tool:
{ "tool_type": "remote", "name": "code_interpreter", "display_name": "Code Interpreter", "description": "Remote MCP server for code execution", "risk_level": "high", "server_url": "https://mcp.example.com/streamable-http", "transport_preference": "auto", "static_headers": { "X-API-Key": "sk-abc123" }, "requires_per_user_auth": false }List tools with filters:
GET /api/v1/tools?tool_type=remote&risk_level=high&is_active=true&q=interpreter&page=1&page_size=20 | Method | Endpoint | Description | Auth Required |
|---|---|---|---|
GET | /api/v1/audit-events | List audit events (with filters) | Admin |
Query parameters: actor_user_id, action, target_id, outcome, from_timestamp, to_timestamp, page, page_size
Supported actions: auth.login, auth.refresh, auth.elevate, tool.create, tool.update, tool.activate, tool.deactivate, tool.delete, tool.remote.test_connection, tool.remote.reprobe
The system uses a dual-token strategy:
| Token | Lifetime | Purpose |
|---|---|---|
| Access token | 15 minutes | Sent as Authorization: Bearer <token> on every API call |
| Refresh token | 14 days | Exchanged for a fresh token pair without re-entering credentials |
| Elevated token | 5 minutes | Short-lived access token for destructive operations |
Access tokens are stateless JWTs. Refresh tokens are additionally tracked in the database and can be revoked (rotation happens on every refresh — the old token is invalidated).
Certain destructive operations (e.g., deleting a tool) require step-up authentication. Even with a valid admin token, the user must re-enter their password via POST /api/v1/auth/elevate to receive a short-lived elevated token. This limits the blast radius if a regular admin token is compromised.
| Operation | Regular User | Admin | Elevated Admin |
|---|---|---|---|
| Login / Refresh | Yes | Yes | Yes |
| View tools | Yes | Yes | Yes |
| Create / Update tools | No | Yes | Yes |
| Activate / Deactivate tools | No | Yes | Yes |
| Delete tools | No | No | Yes |
| Test remote connectivity | No | Yes | Yes |
| Re-probe remote tools | No | Yes | Yes |
| View audit events | No | Yes | Yes |
Function tools represent locally-executed capabilities defined by a JSON Schema. The schema describes the input parameters the tool accepts. When creating a function tool, the input_schema is validated against the JSON Schema specification.
Example input schema:
{ "type": "object", "properties": { "city": { "type": "string", "description": "City name" }, "units": { "type": "string", "enum": ["celsius", "fahrenheit"] } }, "required": ["city"] }Remote tools point to external HTTP servers (typically MCP servers). They store:
- Server URL — the endpoint to reach
- Transport preference —
auto,streamable_http, orsse - Static headers — encrypted at rest (e.g., API keys)
- Per-user auth config — whether end-users must provide their own credentials
- Probe state — the result of the last connectivity check
Remote tools are probed on creation — the server must respond correctly before the tool is registered. If the probe fails, creation is rejected.
The ProbeService verifies remote tool connectivity by sending HTTP GET requests and inspecting responses:
| Scenario | Result |
|---|---|
Server responds with 200 and matching Content-Type | Passed (no auth required) |
Server responds with 401 or 403 | Passed (auth required detected) |
| Server responds with other 4xx/5xx | Failed |
| Connection timeout or network error | Failed |
| Content-Type doesn't match transport | Tries next transport (if auto) |
When transport_preference is auto, the system tries SSE first, then Streamable HTTP.
Probe results include:
detected_transport— which protocol the server speaksrequires_auth— whether the server returned 401/403response_time_ms— latency measurementprobe_valid_until— probes go "stale" after 24 hours (configurable)
| Concern | Implementation |
|---|---|
| Password storage | Argon2id hashing (resistant to GPU/ASIC brute-force) |
| Token signing | HMAC-SHA256 JWTs with configurable secret |
| Token rotation | Refresh tokens are revoked on use (one-time use) |
| Secrets at rest | Remote tool headers encrypted with Fernet (AES-128-CBC) |
| Step-up auth | Destructive operations require re-authentication |
| Rate limiting | Login: 5/min per IP, 20/15min per username. Probes: 10/min per user |
| Audit trail | Every auth and tool operation logged with IP, user-agent, request ID |
| Input validation | Pydantic enforces schemas; tool names match ^[A-Za-z][A-Za-z0-9_]*$ |
| Request tracking | Every request tagged with a unique X-Request-Id header |
┌──────────┐ ┌────────────────┐ │ users │──1:N──▶│ refresh_tokens │ └──────────┘ └────────────────┘ │ 1:N ▼ ┌──────────┐ ┌──────────────────────┐ │ tools │──1:1──▶│ function_tool_configs │ └──────────┘ └──────────────────────┘ │ 1:1 ┌──────────────────────┐ └─────────────▶│ remote_tool_configs │──1:N──▶ remote_tool_connection_tests └──────────────────────┘ │ 1:N ▼ ┌──────────────┐ │ audit_events │ └──────────────┘ All models inherit UUIDPrimaryKeyMixin (UUID v4 primary keys) and TimestampMixin (auto-managed created_at / updated_at columns). Usernames and tool names are stored with a _normalized (lowercased) column for case-insensitive uniqueness.
Database migrations are managed by Alembic:
# Run all pending migrations alembic upgrade head # Create a new migration after model changes alembic revision --autogenerate -m "description of change" # Downgrade one step alembic downgrade -1Note: On first startup with
AUTO_CREATE_SCHEMA=true(the default), the schema is created automatically without needing to run migrations manually.
# Run all tests with coverage pytest # Run only unit tests pytest tests/unit/ # Run only integration tests pytest tests/integration/ # Run with verbose output pytest -v --tb=shortThe test suite uses:
- In-memory SQLite database per test session
TestClientfrom FastAPI for integration tests- Mock HTTP transport for probe service tests (no real network calls)
- Automatic admin user fixture for authenticated endpoint tests
A comprehensive bash-based smoke test that exercises the full API against a running server:
# Start the server first uvicorn app.main:app --port 8000 & # Run the smoke test bash scripts/e2e_smoke_test.shThe smoke test covers: health check, login, token refresh, elevation, function tool CRUD, remote tool creation with probing, connection testing, reprobing, update rollback on probe failure, delete with elevated auth, and audit log verification.
| Scope | Default Limit | Key |
|---|---|---|
| Login (per IP) | 5 requests / minute | Client IP address |
| Login (per username) | 20 requests / 15 minutes | Lowercase username |
| Probe / Test-connection (per user) | 10 requests / minute | User ID |
| Probe / Test-connection (per IP) | 30 requests / minute | Client IP address |
Rate limits use a fixed-window strategy with in-memory storage by default. Override limits via environment variables (see below). When a limit is hit, the API returns 429 Too Many Requests.
| Variable | Default | Description |
|---|---|---|
APP_NAME | Tool Registry Backend | Application display name |
ENVIRONMENT | development | development, test, or production |
DATABASE_URL | sqlite+pysqlite:///./tool_registry.db | SQLAlchemy database connection string |
JWT_SECRET_KEY | replace-this-in-production | Secret key for signing JWT tokens |
JWT_ALGORITHM | HS256 | JWT signing algorithm |
ACCESS_TOKEN_TTL_MINUTES | 15 | Access token lifetime |
REFRESH_TOKEN_TTL_DAYS | 14 | Refresh token lifetime |
ELEVATED_TOKEN_TTL_MINUTES | 5 | Elevated admin token lifetime |
ENCRYPTION_SECRET | replace-this-too | Secret for Fernet encryption of stored headers |
BOOTSTRAP_ADMIN_USERNAME | (none) | Auto-create admin user on startup |
BOOTSTRAP_ADMIN_PASSWORD | (none) | Password for the bootstrap admin |
BOOTSTRAP_ADMIN_ROLE | admin | Role for the bootstrap user (admin or user) |
AUTO_CREATE_SCHEMA | true | Auto-create database tables on startup |
PROBE_TIMEOUT_SECONDS | 5.0 | HTTP timeout for remote tool probes |
PROBE_STALE_AFTER_HOURS | 24 | Hours before a probe result is considered stale |
RATE_LIMIT_STORAGE_URI | memory:// | Rate limiter backend URI |
LOGIN_RATE_LIMIT_IP | 5/minute | Login rate limit per IP |
LOGIN_RATE_LIMIT_USERNAME | 20/15 minutes | Login rate limit per username |
PROBE_RATE_LIMIT_USER | 10/minute | Probe rate limit per user |
PROBE_RATE_LIMIT_IP | 30/minute | Probe rate limit per IP |
This project is proprietary. All rights reserved.