ok

app/api/server.py

@@ -89,8 +89,33 @@ def create_application() -> FastAPI:
     app.include_router(api_v1_router, prefix="/api/v1")
 
     @app.get("/", include_in_schema=False)
-    async def root():
-        return {"message": f"{_settings.app_name} v{_settings.app_version} is running"}
+    async def root(request: Request):
+        from collections import defaultdict
+        routes_by_tag: dict[str, list[dict]] = defaultdict(list)
+        for route in app.routes:
+            if not hasattr(route, "methods") or not hasattr(route, "path"):
+                continue
+            if route.path in ("/", "/health", "/ping", "/openapi.json", "/docs", "/redoc", "/docs/oauth2-redirect"):
+                continue
+            tags = getattr(route, "tags", None) or ["default"]
+            for tag in tags:
+                routes_by_tag[tag].append({
+                    "method": list(route.methods - {"HEAD", "OPTIONS"}),
+                    "path": route.path,
+                    "summary": getattr(route, "summary", ""),
+                })
+        return {
+            "name": _settings.app_name,
+            "version": _settings.app_version,
+            "docs": {
+                "swagger": str(request.base_url) + "docs",
+                "redoc": str(request.base_url) + "redoc",
+            },
+            "endpoints": [
+                {"tag": tag, "routes": sorted(routes, key=lambda r: r["path"])}
+                for tag, routes in sorted(routes_by_tag.items())
+            ],
+        }
 
     @app.get("/health", include_in_schema=False)
     async def root_health():

app/api/v1/router.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, web_search
+from app.api.v1 import batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, web_search
 from app.api.verify import router as verify_router
 
 api_v1_router = APIRouter()
@@ -19,4 +19,5 @@ api_v1_router.include_router(web_search.router, tags=["Web Search"])
 api_v1_router.include_router(sql_validator.router, tags=["SQL Validator"])
 api_v1_router.include_router(semantic_router.router, tags=["Semantic Router"])
 api_v1_router.include_router(token_counter.router, tags=["Token Counter"])
+api_v1_router.include_router(token_generator.router, tags=["Token Generator"])
 api_v1_router.include_router(chat.router, tags=["Chat"])

app/api/v1/token_generator.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import time
+
+from fastapi import APIRouter, Depends
+
+from app.api.deps import require_auth
+from app.models.schemas import (
+    TokenGenerateRequest,
+    TokenGenerateResponse,
+    TokenValidateRequest,
+    TokenValidateResponse,
+)
+from app.services.jwt_service import JWTService
+
+router = APIRouter()
+_service = JWTService()
+
+
+@router.post(
+    "/token/generate",
+    response_model=TokenGenerateResponse,
+    summary="Generate a signed JWT token with custom claims",
+)
+async def generate_token(
+    body: TokenGenerateRequest,
+    auth: str = Depends(require_auth),
+) -> TokenGenerateResponse:
+    start = time.perf_counter()
+    try:
+        result = _service.generate(
+            subject=body.subject,
+            role=body.role,
+            permissions=body.permissions,
+            issuer=body.issuer,
+            audience=body.audience,
+            expiry_minutes=body.expiry_minutes,
+            not_before_minutes=body.not_before_minutes,
+            extra_claims=body.extra_claims,
+            secret=body.secret,
+            algorithm=body.algorithm,
+        )
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
+        return TokenGenerateResponse(
+            success=True,
+            time_ms=elapsed_ms,
+            token=result["token"],
+            claims=result["claims"],
+            expires_at=result["expires_at"],
+            valid_for=result.get("valid_for"),
+            secret=result.get("secret"),
+            algorithm=result["algorithm"],
+        )
+    except Exception as exc:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
+        return TokenGenerateResponse(
+            success=False,
+            time_ms=elapsed_ms,
+            error=str(exc),
+        )
+
+
+@router.post(
+    "/token/validate",
+    response_model=TokenValidateResponse,
+    summary="Validate a JWT token and return its claims",
+)
+async def validate_token(
+    body: TokenValidateRequest,
+    auth: str = Depends(require_auth),
+) -> TokenValidateResponse:
+    start = time.perf_counter()
+    try:
+        result = _service.validate(body.token, body.audience, body.secret, body.algorithm)
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
+        return TokenValidateResponse(
+            success=True,
+            time_ms=elapsed_ms,
+            valid=result["valid"],
+            claims=result["claims"],
+            error=result["error"],
+        )
+    except Exception as exc:
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 3)
+        return TokenValidateResponse(
+            success=False,
+            time_ms=elapsed_ms,
+            valid=False,
+            error=str(exc),
+        )

app/config.py

@@ -61,6 +61,11 @@ class Settings(BaseSettings):
     key_lock_ttl: int = 600
     rounds_per_model: int = 50
 
+    jwt_secret_key: str = "changeme-jwt-secret"
+    jwt_algorithm: str = "HS256"
+    jwt_default_expiry_minutes: int = 30
+    jwt_issuer: str = "all-api-collection"
+
     @property
     def max_upload_mb(self) -> int:
         return self.max_upload_bytes // (1024 * 1024)

app/models/schemas.py

@@ -468,6 +468,46 @@ class WebSearchEngineDescriptionsResponse(BaseModel):
     error: Optional[str] = None
 
 
+class TokenGenerateRequest(BaseModel):
+    subject: str = Field(..., min_length=1, max_length=256, description="Token subject (user ID, client ID, etc.)")
+    role: Optional[str] = Field(None, max_length=64, description="User role for RBAC")
+    permissions: Optional[List[str]] = Field(None, description="List of permission strings")
+    issuer: Optional[str] = Field(None, max_length=128, description="Token issuer (overrides default)")
+    audience: Optional[str] = Field(None, max_length=128, description="Token audience")
+    expiry_minutes: Optional[int] = Field(None, ge=1, le=525600, description="Token lifetime in minutes")
+    not_before_minutes: int = Field(default=0, ge=0, le=525600, description="Delay token validity by N minutes")
+    extra_claims: Optional[Dict[str, Any]] = Field(None, description="Additional custom claims")
+    secret: Optional[str] = Field(None, description="JWT signing secret (auto-generated if not provided)")
+    algorithm: Optional[str] = Field(None, pattern="^(HS256|HS384|HS512)$", description="JWT signing algorithm (defaults to HS256)")
+
+
+class TokenGenerateResponse(BaseModel):
+    success: bool
+    time_ms: float
+    token: Optional[str] = None
+    claims: Optional[Dict[str, Any]] = None
+    expires_at: Optional[str] = None
+    valid_for: Optional[str] = None
+    secret: Optional[str] = None
+    algorithm: Optional[str] = None
+    error: Optional[str] = None
+
+
+class TokenValidateRequest(BaseModel):
+    token: str = Field(..., min_length=1, description="JWT token string to validate")
+    audience: Optional[str] = Field(None, description="Expected audience to verify against")
+    secret: Optional[str] = Field(None, description="Signing secret used during generation (uses server default if not provided)")
+    algorithm: Optional[str] = Field(None, pattern="^(HS256|HS384|HS512)$", description="Algorithm used during generation (uses server default if not provided)")
+
+
+class TokenValidateResponse(BaseModel):
+    success: bool
+    time_ms: float
+    valid: bool
+    claims: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None
+
+
 class SqlValidationRequest(BaseModel):
     query: str = Field(..., min_length=1, max_length=100000, description="SQL query string to validate")
     dialect: Optional[str] = Field(None, description="SQL dialect (mysql, postgres, bigquery, snowflake, sqlite, etc.)")

app/services/jwt_service.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import secrets
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+
+import jwt
+from jwt import PyJWTError
+
+from app.config import get_settings
+
+_SUPPORTED_ALGORITHMS = frozenset({"HS256", "HS384", "HS512"})
+
+
+class JWTService:
+    def __init__(self) -> None:
+        self._settings = get_settings()
+
+    @staticmethod
+    def _generate_secret() -> str:
+        return secrets.token_hex(32)
+
+    def generate(
+        self,
+        *,
+        subject: str,
+        role: Optional[str] = None,
+        permissions: Optional[list[str]] = None,
+        issuer: Optional[str] = None,
+        audience: Optional[str] = None,
+        expiry_minutes: Optional[int] = None,
+        not_before_minutes: int = 0,
+        extra_claims: Optional[dict[str, Any]] = None,
+        secret: Optional[str] = None,
+        algorithm: Optional[str] = None,
+    ) -> dict[str, Any]:
+        now = datetime.now(timezone.utc)
+        expire_min = expiry_minutes if expiry_minutes is not None else self._settings.jwt_default_expiry_minutes
+
+        payload: dict[str, Any] = {
+            "sub": subject,
+            "iat": now,
+            "exp": now + timedelta(minutes=expire_min),
+            "jti": uuid.uuid4().hex,
+        }
+
+        if issuer:
+            payload["iss"] = issuer
+        else:
+            payload["iss"] = self._settings.jwt_issuer
+
+        if audience:
+            payload["aud"] = audience
+
+        if role:
+            payload["role"] = role
+
+        if permissions:
+            payload["permissions"] = permissions
+
+        if not_before_minutes > 0:
+            payload["nbf"] = now + timedelta(minutes=not_before_minutes)
+
+        if extra_claims:
+            payload.update(extra_claims)
+
+        used_secret = secret if secret else self._generate_secret()
+        used_algorithm = algorithm if algorithm else self._settings.jwt_algorithm
+
+        if used_algorithm not in _SUPPORTED_ALGORITHMS:
+            raise ValueError(f"Unsupported algorithm '{used_algorithm}'. Supported: {sorted(_SUPPORTED_ALGORITHMS)}")
+
+        token = jwt.encode(payload, used_secret, algorithm=used_algorithm)
+
+        valid_minutes = expire_min
+        if valid_minutes < 60:
+            valid_for = f"{valid_minutes} minute{'s' if valid_minutes != 1 else ''}"
+        elif valid_minutes < 1440:
+            hours = valid_minutes // 60
+            mins = valid_minutes % 60
+            valid_for = f"{hours} hour{'s' if hours != 1 else ''}"
+            if mins:
+                valid_for += f" {mins} minute{'s' if mins != 1 else ''}"
+        else:
+            days = valid_minutes // 1440
+            hrs = (valid_minutes % 1440) // 60
+            valid_for = f"{days} day{'s' if days != 1 else ''}"
+            if hrs:
+                valid_for += f" {hrs} hour{'s' if hrs != 1 else ''}"
+
+        return {
+            "token": token,
+            "claims": {
+                "sub": subject,
+                "iss": payload["iss"],
+                "aud": audience,
+                "role": role,
+                "permissions": permissions,
+                "jti": payload["jti"],
+                "iat": payload["iat"].isoformat(),
+                "exp": payload["exp"].isoformat(),
+                "nbf": payload.get("nbf").isoformat() if payload.get("nbf") else None,
+            },
+            "expires_at": payload["exp"].isoformat(),
+            "valid_for": valid_for,
+            "secret": used_secret,
+            "algorithm": used_algorithm,
+        }
+
+    def validate(
+        self,
+        token: str,
+        audience: Optional[str] = None,
+        secret: Optional[str] = None,
+        algorithm: Optional[str] = None,
+    ) -> dict[str, Any]:
+        used_secret = secret if secret else self._settings.jwt_secret_key
+        used_algorithms = [algorithm] if algorithm else [self._settings.jwt_algorithm]
+
+        try:
+            payload = jwt.decode(
+                token,
+                used_secret,
+                algorithms=used_algorithms,
+                audience=audience,
+                issuer=self._settings.jwt_issuer,
+                options={
+                    "require": ["sub", "exp", "iat", "jti"],
+                    "verify_signature": True,
+                    "verify_exp": True,
+                    "verify_iat": True,
+                    "verify_aud": audience is not None,
+                    "verify_iss": True,
+                },
+            )
+            return {
+                "valid": True,
+                "claims": {
+                    "sub": payload.get("sub"),
+                    "iss": payload.get("iss"),
+                    "aud": payload.get("aud"),
+                    "role": payload.get("role"),
+                    "permissions": payload.get("permissions"),
+                    "jti": payload.get("jti"),
+                    "iat": datetime.fromtimestamp(payload["iat"], tz=timezone.utc).isoformat() if payload.get("iat") else None,
+                    "exp": datetime.fromtimestamp(payload["exp"], tz=timezone.utc).isoformat() if payload.get("exp") else None,
+                    "nbf": datetime.fromtimestamp(payload["nbf"], tz=timezone.utc).isoformat() if payload.get("nbf") else None,
+                },
+                "error": None,
+            }
+        except jwt.ExpiredSignatureError:
+            return {"valid": False, "claims": None, "error": "Token has expired."}
+        except jwt.InvalidAudienceError:
+            return {"valid": False, "claims": None, "error": "Token audience does not match."}
+        except jwt.InvalidIssuerError:
+            return {"valid": False, "claims": None, "error": "Token issuer does not match."}
+        except jwt.InvalidTokenError as exc:
+            return {"valid": False, "claims": None, "error": f"Invalid token: {exc}"}
+        except PyJWTError as exc:
+            return {"valid": False, "claims": None, "error": f"JWT validation error: {exc}"}

requirements.txt

@@ -25,6 +25,7 @@ spacy>=3.7.0
 phonenumbers>=8.13.0
 sqlglot>=20.0.0
 tiktoken>=0.9.0
+PyJWT>=2.9.0
 
 # Async database drivers
 aiomysql>=0.3.2