API Validation Gateway

A gateway that validates API endpoint responses through a sequential pipeline of specialized AI agents before granting production access. Built with FastAPI, Temporal, and OpenAI.

Motivation

When onboarding third-party integrations or internal services, validating that API responses conform to expected contracts is tedious and error-prone when done manually. This project automates that process: the submitter sends their endpoint responses one by one, each is reviewed by a dedicated AI agent, and a human admin makes the final call. Access credentials are only issued after the full pipeline passes.

How It Works

User submits auth response
        │
        ▼
┌─────────────────────┐
│  Auth Validator     │  Checks token structure, required fields, expiration
└────────┬────────────┘
         │ pass
         ▼
┌─────────────────────┐
│  Data Validator     │  Checks data structure, types, pagination, nulls
└────────┬────────────┘
         │ pass
         ▼
┌─────────────────────┐
│  Health Validator   │  Checks status, version, metrics, error indicators
└────────┬────────────┘
         │ pass
         ▼
┌─────────────────────┐
│  Human Review       │  Admin approves or rejects with comments
└────────┬────────────┘
         │ approved
         ▼
  Production credentials issued

Each stage waits for the user to submit the corresponding endpoint response, then an AI agent validates it using OpenAI function calling. If any stage fails, the workflow is rejected immediately. The workflow state is durable — it survives restarts thanks to Temporal.

Tech Stack

Layer	Technology
API	FastAPI
Workflow orchestration	Temporal
AI validation	OpenAI (function calling)
Resilience	Circuit breaker pattern
Configuration	Externalized YAML prompt configs

Quick Start

Prerequisites

• Python 3.11+
• Docker and Docker Compose
• OpenAI API key

Setup

# Clone and install
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt

# Configure environment
cp .env.example .env
# Edit .env with your OPENAI_API_KEY

# Start Temporal
docker-compose up -d temporal temporal-ui postgresql

# Terminal 1 — start the Temporal worker
python -m app.temporal.worker

# Terminal 2 — start the API
uvicorn app.main:app --reload --port 8000

• API docs: http://localhost:8000/docs
• Temporal UI: http://localhost:8080

API Reference

Validation Flow

Method	Endpoint	Description
POST	/api/v1/validate	Submit auth response — starts a new workflow
POST	/api/v1/validate/{workflow_id}	Submit data or health response to existing workflow
GET	/api/v1/validate/{workflow_id}/status	Poll workflow status and agent results

Admin

Method	Endpoint	Description
GET	/api/v1/admin/pending-approvals	List workflows awaiting human review
GET	/api/v1/admin/workflow/{workflow_id}	Full workflow details for review
POST	/api/v1/admin/approve/{workflow_id}	Approve or reject

Usage Example

# 1. Submit auth endpoint response (creates workflow)
curl -X POST http://localhost:8000/api/v1/validate \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-123",
    "endpoint_response": {
      "endpoint_type": "auth",
      "content": {"token": "eyJhbGci...", "expires_in": 3600, "token_type": "Bearer"}
    }
  }'
# -> returns workflow_id

# 2. Submit data endpoint response
curl -X POST http://localhost:8000/api/v1/validate/{workflow_id} \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-123",
    "endpoint_response": {
      "endpoint_type": "data",
      "content": {"items": [...], "total": 42, "page": 1, "per_page": 10}
    }
  }'

# 3. Submit health endpoint response
curl -X POST http://localhost:8000/api/v1/validate/{workflow_id} \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-123",
    "endpoint_response": {
      "endpoint_type": "health",
      "content": {"status": "healthy", "version": "1.2.0", "uptime": 99.9}
    }
  }'

# 4. Poll status
curl http://localhost:8000/api/v1/validate/{workflow_id}/status

# 5. Admin approval (after all agents pass)
curl -X POST http://localhost:8000/api/v1/admin/approve/{workflow_id} \
  -H "Content-Type: application/json" \
  -d '{"decision": "approved", "comments": "All checks passed."}'

Project Structure

app/
├── api/
│   ├── routes/         # validation.py, admin.py
│   └── models/         # requests.py, responses.py
├── temporal/
│   ├── workflows/      # EndpointValidationWorkflow
│   └── activities/     # ai_validators.py, access_management.py
├── config/
│   └── prompts.py      # YAML prompt loader with Jinja2 templating
├── core/               # Logging, exceptions
└── main.py

config/
└── prompts/
    ├── auth_validator.yaml
    ├── data_validator.yaml
    └── health_validator.yaml

tests/
├── unit/
├── integration/
└── contract/

Configuration

Variable	Description	Default
OPENAI_API_KEY	OpenAI API key	Required
OPENAI_MODEL	Model to use	gpt-4o
TEMPORAL_HOST	Temporal server address	localhost:7233
TEMPORAL_NAMESPACE	Temporal namespace	default
LOG_LEVEL	Logging level	INFO

Running Tests

pytest                          # all tests
pytest tests/unit/              # unit only
pytest tests/integration/       # integration only
pytest tests/contract/          # contract only
pytest --cov=app --cov-report=html

License

MIT