Contract Intelligence Pipeline

python
import httpx
import asyncio

BASE_URL = "https://api.alloovium.com/api/v1"
API_KEY = "ak_live_YOUR_KEY_HERE"

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}


async def create_project(client: httpx.AsyncClient, name: str) -> str:
    """Create a project and return its ID."""
    resp = await client.post(
        f"{BASE_URL}/projects",
        json={"name": name, "project_type": "commercial"},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["id"]

Alloovium uses a prepare/confirm flow for uploads. The prepare step returns signed S3 URLs. You upload directly to S3, then confirm — this keeps your API key out of the multipart boundary and removes the API server from the data path.

python
import os
from pathlib import Path


async def upload_contract(
    client: httpx.AsyncClient,
    project_id: str,
    contract_path: str,
) -> str:
    """Upload a contract PDF and return the document ID."""
    path = Path(contract_path)
    filename = path.name
    file_size = path.stat().st_size

    # Step 1a — prepare: request a signed upload URL
    resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-prepare",
        json={
            "documents": [
                {
                    "filename": filename,
                    "content_type": "application/pdf",
                    "size": file_size,
                }
            ]
        },
        headers=HEADERS,
    )
    resp.raise_for_status()
    prepared = resp.json()["documents"][0]
    doc_id = prepared["document_id"]
    upload_url = prepared["upload_url"]
    upload_fields = prepared.get("upload_fields", {})

    # Step 1b — stream the file directly to S3
    with open(contract_path, "rb") as f:
        if upload_fields:
            # Multipart POST to S3
            s3_resp = await client.post(
                upload_url,
                data=upload_fields,
                files={"file": (filename, f, "application/pdf")},
            )
        else:
            # Simple PUT to presigned URL
            s3_resp = await client.put(
                upload_url,
                content=f.read(),
                headers={"Content-Type": "application/pdf"},
            )
    s3_resp.raise_for_status()

    # Step 1c — confirm: tell Alloovium the upload succeeded
    confirm_resp = await client.post(
        f"{BASE_URL}/projects/{project_id}/documents/bulk-confirm",
        json={"document_ids": [doc_id]},
        headers=HEADERS,
    )
    confirm_resp.raise_for_status()

    return doc_id

python
async def create_conversation(
    client: httpx.AsyncClient,
    project_id: str,
) -> str:
    """Create a conversation scoped to the project and return its ID."""
    resp = await client.post(
        f"{BASE_URL}/conversations",
        json={"project_id": project_id, "title": "Contract extraction"},
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["id"]

python
async def ask_and_collect(
    client: httpx.AsyncClient,
    conversation_id: str,
    question: str,
) -> str:
    """Send a message and collect the full streamed answer as a string."""
    full_answer: list[str] = []

    async with client.stream(
        "POST",
        f"{BASE_URL}/conversations/{conversation_id}/messages",
        json={"content": question},
        headers={**HEADERS, "Accept": "text/event-stream"},
        timeout=120.0,
    ) as stream:
        async for line in stream.aiter_lines():
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload in ("", "[DONE]"):
                continue

            import json
            try:
                event = json.loads(payload)
            except json.JSONDecodeError:
                continue

            event_type = event.get("type")
            if event_type == "token":
                full_answer.append(event.get("token", ""))
            elif event_type == "complete":
                # complete event contains the full answer too
                return event.get("answer", "".join(full_answer))

    return "".join(full_answer)

Use structured prompts that ask for a specific format. Asking for JSON output in the prompt makes parsing reliable. The chat API grounds its answers in the uploaded documents, so these prompts return values drawn directly from the contract text.

python
OBLIGATION_PROMPT = """
List all contractor obligations from this contract.
For each obligation return a JSON object with these fields:
- clause_ref: the clause number (e.g. "12.3")
- description: a one-sentence plain-language description
- party: who is obligated ("contractor", "principal", or "both")
- deadline: ISO date if a deadline is stated, otherwise null

Return a JSON array. No markdown, no commentary — raw JSON only.
"""

PAYMENT_TERMS_PROMPT = """
Extract all payment terms from this contract.
Return a JSON object with:
- payment_schedule: array of {milestone, amount_aud, due_date}
- retention_percent: number (e.g. 5 for 5%)
- retention_release_condition: string
- late_payment_interest_percent: number or null
- payment_claim_period_days: number

Raw JSON only.
"""

MILESTONE_PROMPT = """
List all project milestones and completion dates from this contract.
Return a JSON array where each item has:
- name: milestone name
- date: ISO date or null if not stated
- liquidated_damages_per_day_aud: number or null

Raw JSON only.
"""

python
async def push_to_procore(
    intelligence: ContractIntelligence,
    procore_project_id: int,
    procore_token: str,
) -> None:
    """Create Procore schedule items for each extracted milestone."""
    async with httpx.AsyncClient() as client:
        for milestone in intelligence.milestones:
            await client.post(
                f"https://api.procore.com/rest/v1.0/projects/{procore_project_id}/schedule/tasks",
                json={
                    "task": {
                        "name": milestone["name"],
                        "start_date": milestone.get("date"),
                        "end_date": milestone.get("date"),
                        "notes": (
                            f"LD: AUD {milestone['liquidated_damages_per_day_aud']}/day"
                            if milestone.get("liquidated_damages_per_day_aud")
                            else ""
                        ),
                    }
                },
                headers={"Authorization": f"Bearer {procore_token}"},
            )

python
async def push_obligations_to_jira(
    intelligence: ContractIntelligence,
    jira_base_url: str,
    jira_project_key: str,
    jira_token: str,
) -> None:
    """Create a Jira issue for each contractor obligation."""
    async with httpx.AsyncClient() as client:
        for obligation in intelligence.obligations:
            if obligation.get("party") not in ("contractor", "both"):
                continue

            await client.post(
                f"{jira_base_url}/rest/api/3/issue",
                json={
                    "fields": {
                        "project": {"key": jira_project_key},
                        "issuetype": {"name": "Task"},
                        "summary": f"[{obligation['clause_ref']}] {obligation['description']}",
                        "duedate": obligation.get("deadline"),
                        "labels": ["contract-obligation"],
                    }
                },
                headers={
                    "Authorization": f"Bearer {jira_token}",
                    "Content-Type": "application/json",
                },
            )