Skip to content

samith2002/LLM-Tool-Registry

Repository files navigation

Tool Registry Backend

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.


Table of Contents


Architecture

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.

Tech Stack

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+

Project Structure

├── 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 

Getting Started

Prerequisites

  • Python 3.14+
  • pip (or any PEP 517-compatible installer)

Installation

# 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]"

Configuration

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.

Running the Server

# Development mode with hot-reload uvicorn app.main:app --reload --port 8000

The 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

Seeding an Admin User

An admin user can be created in two ways:

  1. Automatic (via environment): Set BOOTSTRAP_ADMIN_USERNAME and BOOTSTRAP_ADMIN_PASSWORD — the user is created on first startup if they don't already exist.

  2. Manual (via CLI):

    python -m app.cli seed-admin --username admin --password "AdminPass123!"

API Reference

Authentication

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" }

Tools

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 

Audit Events

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


Authentication & Authorization

Token Lifecycle

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).

Elevated Access (Step-Up Auth)

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.

Permission Matrix

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

Tool Types

Function Tools

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

Remote tools point to external HTTP servers (typically MCP servers). They store:

  • Server URL — the endpoint to reach
  • Transport preferenceauto, streamable_http, or sse
  • 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.


Remote Tool Probing

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 speaks
  • requires_auth — whether the server returned 401/403
  • response_time_ms — latency measurement
  • probe_valid_until — probes go "stale" after 24 hours (configurable)

Security

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

Database

Schema Overview

┌──────────┐ ┌────────────────┐ │ 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.

Migrations

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 -1

Note: On first startup with AUTO_CREATE_SCHEMA=true (the default), the schema is created automatically without needing to run migrations manually.


Testing

Unit & Integration Tests

# 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=short

The test suite uses:

  • In-memory SQLite database per test session
  • TestClient from FastAPI for integration tests
  • Mock HTTP transport for probe service tests (no real network calls)
  • Automatic admin user fixture for authenticated endpoint tests

End-to-End Smoke Test

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.sh

The 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.


Rate Limiting

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.


Environment Variables

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

License

This project is proprietary. All rights reserved.

About

FastAPI backend for managing an AI-agent tool catalog — JWT auth, RBAC, encrypted secrets, remote MCP server probing, audit logging & rate limiting.

Topics

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors