LexaShield Engine SDK

Build legal processing engines in Python. The SDK provides a FastAPI server with ready-made endpoints, storage, and webhooks so you can focus on your analysis logic.

Installation

• Create and activate a virtual environment (recommended):

  • macOS/Linux: python3 -m venv venv

    source venv/bin/activate

  • Windows: python -m venv venv

    venv\Scripts\activate

• Install the SDK from the local folder:

  • Ensure you are in the root folder of the SDK (the one containing setup.py or pyproject.toml).
  • Run: pip install -e .

Minimal Example

from lexashield_engine import BaseEngine, ProcessingContext

class MyLegalEngine(BaseEngine):
    async def process_query(self, ctx: ProcessingContext, query: str) -> dict:
        await ctx.update_progress(50, "Analyzing query...")
        return {"result": f"Analysis of: {query}"}

    async def process_file(self, ctx: ProcessingContext, file_path: str) -> dict:
        await ctx.update_progress(30, "Reading file...")
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
        await ctx.update_progress(100, "Complete")
        return {"result": f"File size: {len(content)} bytes"}

if __name__ == "__main__":
    MyLegalEngine().run()

Configure

# Required
BACKEND_URL=https://api.lexashield.com

# Optional
ENGINE_HOST=0.0.0.0
ENGINE_PORT=8080
ENGINE_PUBLIC_URL=https://your-engine-public-url.com
LAW_PACK_STORAGE_DIR=/data/law_packs
RESULTS_STORAGE_DIR=/data/results
LOG_LEVEL=INFO
LOG_FILE=./logs/engine.log
LOG_TO_CONSOLE=true

How it works

Your engine exposes endpoints and runs background tasks; the SDK generates result files and sends webhooks to the backend.

Built-in Endpoints

• GET /health
• POST /api/law-packs/distribute (download/register law packs via presigned URLs)
• DELETE /api/law-packs/delete
• POST /api/process (accept query or file; schedules background work)
• GET /api/results/{task_id}/{filename}
• DELETE /api/results/{task_id}/cleanup

What You Implement

• process_query(ctx: ProcessingContext, query: str) -> dict
• process_file(ctx: ProcessingContext, file_path: str) -> dict

Return a dict that includes at least {"result": str}. The SDK turns it into result.html and serves it.

Hooks (optional overrides)

• on_startup()
• on_shutdown()
• on_law_pack_received(law_pack_id: int, file_paths: List[Path])
• on_law_pack_deleted(law_pack_id: int)

ProcessingContext

• task_id: str, law_pack_id: int
• await ctx.update_progress(percent: int, message: Optional[str])
• ctx.elapsed_seconds: float

Configuration (env or .env)

• BACKEND_URL (required)
• ENGINE_HOST, ENGINE_PORT, ENGINE_PUBLIC_URL
• LAW_PACK_STORAGE_DIR, RESULTS_STORAGE_DIR
• LOG_LEVEL, LOG_FILE, LOG_TO_CONSOLE

Customize

• Override _generate_html_result(self, result_data: Dict[str, Any]) -> str to control the HTML layout/style.
• Manually register packs in tests: engine.register_law_pack(1, [Path("pack1.jsonl"), Path("pack2.jsonl")]).

CLI and Testing

# Scaffolding (create a ready-to-edit engine project)
lexashield-engine create my-legal-engine

# Mock backend (receives webhooks)
lexashield-mock-backend --port 8000

# Validate your running engine
lexashield-validate-engine --url http://localhost:8080