Discovery Pipeline

Architecture

Prerequisites

Case.dev API key
Node.js 18+ or Python 3.9+
Documents to process (PDFs, images, Word docs)

Step 1: Create a vault

casedev vault create --name "Matter 2024-1234 - Discovery"

import Casedev from 'casedev';
import fs from 'fs';
import path from 'path';

const client = new Casedev({ apiKey: process.env.CASEDEV_API_KEY });

const vault = await client.vault.create({
  name: 'Matter 2024-1234 - Discovery',
  description: 'Documents received from opposing counsel'
});

console.log(`Created vault: ${vault.id}`);

import os
import casedev

client = casedev.Casedev(api_key=os.environ['CASEDEV_API_KEY'])

vault = client.vault.create(
    name='Matter 2024-1234 - Discovery',
    description='Documents received from opposing counsel'
)

print(f'Created vault: {vault.id}')

vault, _ := client.Vault.New(ctx, casedev.VaultNewParams{
	Name: casedev.F("Matter 2024-1234 - Discovery"),
	Description: casedev.F("Documents received from opposing counsel"),
})
fmt.Println(vault.ID)

Use webhooks to get notified when documents finish processing instead of polling.

casedev vault ingest --id $VAULT_ID --object-id $OBJECT_ID

const subscription = await client.vault.events.subscriptions.create(vault.id, {
  callbackUrl: 'https://your-app.com/webhooks/case-vault',
  eventTypes: ['vault.ingest.completed', 'vault.ingest.failed']
});

console.log(`Webhook subscription: ${subscription.id}`);

subscription = client.vault.events.subscriptions.create(vault.id,
    callback_url='https://your-app.com/webhooks/case-vault',
    event_types=['vault.ingest.completed', 'vault.ingest.failed']
)

print(f'Webhook subscription: {subscription.id}')

result, _ := client.Vault.Ingest(ctx, objectID, casedev.VaultIngestParams{
	ID: casedev.F(vaultID),
})
fmt.Println(result.Status)

Your webhook endpoint receives events like:

Webhook payload

{
  "id": "evt_abc123",
  "eventType": "vault.ingest.completed",
  "vaultId": "vault_abc123",
  "objectId": "obj_xyz789",
  "data": { "status": "completed" }
}

Webhook delivery is at-least-once. Use the id field as an idempotency key and design handlers to safely process duplicates. See Vault Webhooks for signing and retry details.

Step 3: Batch upload documents

casedev vault upload \
  --id $VAULT_ID \
  --filename "document.pdf" \
  --content-type "application/pdf"

async function uploadDocuments(vaultId: string, documentsDir: string) {
  const files = fs.readdirSync(documentsDir);
  const results = [];
  
  const contentTypes: Record<string, string> = {
    '.pdf': 'application/pdf',
    '.doc': 'application/msword',
    '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
    '.jpg': 'image/jpeg',
    '.jpeg': 'image/jpeg',
    '.png': 'image/png',
    '.tiff': 'image/tiff',
    '.txt': 'text/plain',
  };
  
  for (const file of files) {
    const filePath = path.join(documentsDir, file);
    const stat = fs.statSync(filePath);
    
    if (!stat.isFile()) continue;
    
    const ext = path.extname(file).toLowerCase();
    const contentType = contentTypes[ext] || 'application/octet-stream';
    
    // Get presigned upload URL
    const upload = await client.vault.upload(vaultId, {
      filename: file,
      contentType,
      metadata: {
        source: 'discovery',
        original_path: filePath,
      }
    });
    
    // Upload file to S3
    const fileBuffer = fs.readFileSync(filePath);
    await fetch(upload.uploadUrl, {
      method: 'PUT',
      headers: { 'Content-Type': contentType },
      body: fileBuffer
    });
    
    console.log(`Uploaded: ${file}`);
    results.push({ file, objectId: upload.objectId });
  }
  
  return results;
}

import requests
from pathlib import Path

def upload_documents(vault_id: str, documents_dir: str):
    files = list(Path(documents_dir).iterdir())
    results = []
    
    content_types = {
        '.pdf': 'application/pdf',
        '.doc': 'application/msword',
        '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
        '.jpg': 'image/jpeg',
        '.jpeg': 'image/jpeg',
        '.png': 'image/png',
        '.tiff': 'image/tiff',
        '.txt': 'text/plain',
    }
    
    for file_path in files:
        if not file_path.is_file():
            continue
        
        content_type = content_types.get(file_path.suffix.lower(), 'application/octet-stream')
        
        # Get presigned upload URL
        upload = client.vault.upload(vault_id,
            filename=file_path.name,
            content_type=content_type,
            metadata={
                'source': 'discovery',
                'original_path': str(file_path),
            }
        )
        
        # Upload file to S3
        with open(file_path, 'rb') as f:
            requests.put(upload.upload_url, data=f,
                headers={'Content-Type': content_type})
        
        print(f'Uploaded: {file_path.name}')
        results.append({'file': file_path.name, 'object_id': upload.object_id})
    
    return results

upload, _ := client.Vault.Upload(ctx, vaultID, casedev.VaultUploadParams{
	Filename:    casedev.F("document.pdf"),
	ContentType: casedev.F("application/pdf"),
})
// PUT file to upload.UploadURL via net/http
fmt.Println(upload.ObjectID)

Step 4: Trigger ingestion

Ingestion runs OCR (if needed) and generates embeddings for search. Your webhook will fire when each document finishes.

casedev vault ingest --id $VAULT_ID --object-id $OBJECT_ID

async function ingestDocuments(vaultId: string, uploads: { file: string; objectId: string }[]) {
  for (const { file, objectId } of uploads) {
    await client.vault.ingest(vaultId, objectId);
    console.log(`Ingesting: ${file}`);
  }
  
  // Ingestion runs async — your webhook endpoint receives
  // vault.ingest.completed or vault.ingest.failed for each document
}

def ingest_documents(vault_id: str, uploads: list):
    for upload in uploads:
        client.vault.ingest(upload['object_id'], id=vault_id)
        print(f'Ingesting: {upload["file"]}')
    
    # Ingestion runs async — your webhook endpoint receives
    # vault.ingest.completed or vault.ingest.failed for each document

result, _ := client.Vault.Ingest(ctx, objectID, casedev.VaultIngestParams{
	ID: casedev.F(vaultID),
})
fmt.Println(result.Status)

Step 5: Search your documents

Once your webhook confirms ingestion is complete, documents are searchable.

casedev vault search \
  --id $VAULT_ID \
  --query "search query"

async function searchDiscovery(vaultId: string, query: string) {
  const results = await client.vault.search(vaultId, {
    query,
    method: 'hybrid',  // Combines semantic + keyword
    limit: 10
  });
  
  console.log(`\nResults for: "${query}"\n`);
  
  for (const chunk of results.chunks) {
    console.log(`${chunk.filename} (page ${chunk.page})`);
    console.log(`   Score: ${chunk.hybridScore.toFixed(2)}`);
    console.log(`   "${chunk.text.substring(0, 200)}..."\n`);
  }
  
  return results;
}

def search_discovery(vault_id: str, query: str):
    results = client.vault.search(vault_id,
        query=query,
        method='hybrid',  # Combines semantic + keyword
        top_k=10
    )
    
    print(f'\nResults for: "{query}"\n')
    
    for chunk in results.chunks:
        print(f'{chunk.filename} (page {chunk.page})')
        print(f'   Score: {chunk.hybrid_score:.2f}')
        print(f'   "{chunk.text[:200]}..."\n')
    
    return results

results, _ := client.Vault.Search(ctx, vaultID, casedev.VaultSearchParams{
	Query: casedev.F("search query"),
	Method: casedev.F(casedev.VaultSearchParamsMethodHybrid),
})
for _, chunk := range results.Chunks {
	fmt.Println(chunk.Text)
}

Complete example

casedev vault create --name "Matter 2024-1234 - Discovery"

import Casedev from 'casedev';
import fs from 'fs';
import path from 'path';

const client = new Casedev({ apiKey: process.env.CASEDEV_API_KEY });

async function main() {
  const documentsDir = './discovery_dump';
  
  // 1. Create vault
  const vault = await client.vault.create({
    name: 'Matter 2024-1234 - Discovery',
    description: 'Documents from opposing counsel'
  });
  
  // 2. Subscribe to ingestion events
  await client.vault.events.subscriptions.create(vault.id, {
    callbackUrl: 'https://your-app.com/webhooks/case-vault',
    eventTypes: ['vault.ingest.completed', 'vault.ingest.failed']
  });
  
  // 3. Upload and ingest all documents
  const files = fs.readdirSync(documentsDir);
  for (const file of files) {
    const filePath = path.join(documentsDir, file);
    if (!fs.statSync(filePath).isFile()) continue;
    
    const upload = await client.vault.upload(vault.id, {
      filename: file,
      contentType: 'application/pdf'
    });
    
    await fetch(upload.uploadUrl, {
      method: 'PUT',
      body: fs.readFileSync(filePath)
    });
    
    await client.vault.ingest(vault.id, upload.objectId);
    console.log(`Queued: ${file}`);
  }
  
  // 4. Search (after webhook confirms ingestion is complete)
  const results = await client.vault.search(vault.id, {
    query: 'evidence of safety violations in 2023',
    method: 'hybrid',
    limit: 10
  });
  
  console.log(results.chunks);
}

main();

import os
import casedev
import requests
from pathlib import Path

client = casedev.Casedev(api_key=os.environ['CASEDEV_API_KEY'])

def main():
    documents_dir = './discovery_dump'
    
    # 1. Create vault
    vault = client.vault.create(
        name='Matter 2024-1234 - Discovery',
        description='Documents from opposing counsel'
    )
    
    # 2. Subscribe to ingestion events
    client.vault.events.subscriptions.create(vault.id,
        callback_url='https://your-app.com/webhooks/case-vault',
        event_types=['vault.ingest.completed', 'vault.ingest.failed']
    )
    
    # 3. Upload and ingest all documents
    for file_path in Path(documents_dir).iterdir():
        if not file_path.is_file():
            continue
        
        upload = client.vault.upload(vault.id,
            filename=file_path.name,
            content_type='application/pdf'
        )
        
        with open(file_path, 'rb') as f:
            requests.put(upload.upload_url, data=f)
        
        client.vault.ingest(upload.object_id, id=vault.id)
        print(f'Queued: {file_path.name}')
    
    # 4. Search (after webhook confirms ingestion is complete)
    results = client.vault.search(vault.id,
        query='evidence of safety violations in 2023',
        method='hybrid',
        top_k=10
    )
    
    print(results.chunks)

if __name__ == '__main__':
    main()

vault, _ := client.Vault.New(ctx, casedev.VaultNewParams{
	Name: casedev.F("Matter 2024-1234 - Discovery"),
	Description: casedev.F("Documents from opposing counsel"),
})
fmt.Println(vault.ID)

Production tip: For large document sets (1000+), use parallel uploads with a concurrency limit of 10-20 to maximize throughput while avoiding rate limits.

​Architecture

​Prerequisites

​Step 1: Create a vault

​Step 2: Subscribe to ingestion events

​Step 3: Batch upload documents

​Step 4: Trigger ingestion

​Step 5: Search your documents

​Complete example

Architecture

Prerequisites

Step 1: Create a vault

Step 2: Subscribe to ingestion events

Step 3: Batch upload documents

Step 4: Trigger ingestion

Step 5: Search your documents

Complete example