> ## Documentation Index
> Fetch the complete documentation index at: https://docs.graphorlm.com/llms.txt
> Use this file to discover all available pages before exploring further.
# SDK Overview
> Complete overview of the Graphor SDKs for Python and TypeScript: Sources, Chat, Extraction, and Retrieval
The Graphor SDKs provide convenient access to the Graphor REST API from Python and TypeScript/JavaScript applications. Both libraries include type definitions for all request params and response fields.
The Python SDK supports Python 3.9+ and offers both synchronous and asynchronous clients.
The TypeScript SDK provides full type safety and works in Node.js, Deno, Bun, and modern browsers.
This page provides a comprehensive overview of the SDK. It covers the full lifecycle:
1. **Data Ingestion (Sources)**: Ingest, poll status, list, get elements, and manage documents by file\_id
2. **Document Chat**: Ask questions about your documents with conversational memory
3. **Data Extraction**: Extract structured data using JSON Schema
4. **Prebuilt Retrieval**: Retrieve relevant document chunks via semantic search
View the Python SDK source code, report issues, and contribute.
View the TypeScript SDK source code, report issues, and contribute.
## Installation
Install the Graphor SDK from PyPI:
```bash theme={null}
pip install graphor
```
Python 3.9 or higher is required.
Install the Graphor SDK from npm:
```bash theme={null}
npm install graphor
```
Or with your preferred package manager:
```bash theme={null}
yarn add graphor
```
```bash theme={null}
pnpm add graphor
```
TypeScript 4.9+ and Node.js 20+ (LTS) are recommended. Also works in Deno v1.28+, Bun 1.0+, Cloudflare Workers, Vercel Edge Runtime, and modern browsers.
## Data Ingestion (Sources)
The Sources methods cover the full ingestion lifecycle:
Ingest documents from files, URLs, GitHub, and YouTube (returns build\_id; poll for file\_id)
Reprocess an existing source with a different partition method
Retrieve all sources with status and metadata
Retrieve structured elements/partitions from processed sources
Permanently remove sources from your project
## Document Chat
Once your data is ingested, use the Chat method to ask questions:
Ask natural language questions about your documents with conversational memory and structured outputs
## Data Extraction
Extract specific structured data from your documents using schemas:
Extract structured information from documents using JSON Schema and natural language instructions
## Prebuilt Retrieval
Retrieve relevant document chunks via semantic search:
Retrieve relevant document chunks using semantic search for custom LLM integration
## What "Data Ingestion" includes
* **Ingest**: Create a new source (file, URL, GitHub, YouTube); returns `build_id`; poll **get build status** until ready, then use `file_id`
* **Reprocess**: Reprocess an existing source with a different partition method (optional)
* **List**: Monitor status and metadata; optionally filter by `file_ids`
* **Get elements**: Retrieve structured elements/partitions by `file_id` after processing
* **Delete**: Remove a source by `file_id`
## Authentication
All SDK methods require authentication using API tokens. You can provide your API key in two ways:
### Environment Variable (Recommended)
Set the `GRAPHOR_API_KEY` environment variable:
```bash theme={null}
export GRAPHOR_API_KEY="grlm_your_api_key_here"
```
```python theme={null}
from graphor import Graphor
# API key is automatically read from GRAPHOR_API_KEY
client = Graphor()
```
```typescript theme={null}
import Graphor from 'graphor';
// API key is automatically read from GRAPHOR_API_KEY
const client = new Graphor();
```
### Direct Initialization
```python theme={null}
from graphor import Graphor
client = Graphor(api_key="grlm_your_api_key_here")
```
```typescript theme={null}
import Graphor from 'graphor';
const client = new Graphor({ apiKey: 'grlm_your_api_key_here' });
```
Learn how to generate and manage API tokens in the [API Tokens guide](/guides/api-tokens).
### Token Security
* **Never expose tokens** in client-side code or public repositories
* **Use environment variables** to store tokens securely
* **Rotate tokens regularly** for enhanced security
* **Use different tokens** for different environments (dev/staging/prod)
## Async Usage
Simply import `AsyncGraphor` instead of `Graphor` and use `await` with each API call:
```python theme={null}
import asyncio
from graphor import AsyncGraphor
client = AsyncGraphor()
async def main():
build_id = await client.sources.ingest_file(file=b"raw file contents")
print(build_id)
asyncio.run(main())
```
All TypeScript SDK methods are async by default and return Promises:
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
const client = new Graphor();
async function main() {
const buildId = await client.sources.ingestFile({ file: fs.createReadStream('document.pdf') });
console.log(buildId);
}
main();
```
## Available Methods
### Sources
| Method | Description |
| ---------------------------------------------------------------------------- | ---------------------------------------------------- |
| [`client.sources.ingest_file()`](/sdk/sources/upload#upload-a-file) | Ingest a local file (returns `build_id`) |
| [`client.sources.ingest_url()`](/sdk/sources/upload#upload-from-url) | Ingest from a web URL |
| [`client.sources.ingest_github()`](/sdk/sources/upload#upload-from-github) | Ingest from GitHub |
| [`client.sources.ingest_youtube()`](/sdk/sources/upload#upload-from-youtube) | Ingest from YouTube |
| [`client.sources.get_build_status()`](/sdk/sources/upload#get-build-status) | Poll build status; returns `file_id` when ready |
| [`client.sources.reprocess()`](/sdk/sources/process) | Reprocess a source by `file_id` (returns `build_id`) |
| [`client.sources.list()`](/sdk/sources/list) | List all sources (optional `file_ids` filter) |
| [`client.sources.get_elements()`](/sdk/sources/list-elements) | Get parsed elements by `file_id` |
| [`client.sources.delete()`](/sdk/sources/delete) | Delete a source by `file_id` |
| Method | Description |
| --------------------------------------------------------------------------- | --------------------------------------------------- |
| [`client.sources.ingestFile()`](/sdk/sources/upload#upload-a-file) | Ingest a local file (returns `build_id`) |
| [`client.sources.ingestUrl()`](/sdk/sources/upload#upload-from-url) | Ingest from a web URL |
| [`client.sources.ingestGitHub()`](/sdk/sources/upload#upload-from-github) | Ingest from GitHub |
| [`client.sources.ingestYoutube()`](/sdk/sources/upload#upload-from-youtube) | Ingest from YouTube |
| [`client.sources.getBuildStatus()`](/sdk/sources/upload#get-build-status) | Poll build status; returns `fileId` when ready |
| [`client.sources.reprocess()`](/sdk/sources/process) | Reprocess a source by `fileId` (returns `build_id`) |
| [`client.sources.list()`](/sdk/sources/list) | List all sources (optional `fileIds` filter) |
| [`client.sources.getElements()`](/sdk/sources/list-elements) | Get parsed elements by `fileId` |
| [`client.sources.delete()`](/sdk/sources/delete) | Delete a source by `fileId` |
### Chat & Extraction
| Method | Description |
| ------------------------------------------------------- | ----------------------------------------- |
| [`client.sources.ask()`](/sdk/chat) | Ask questions about your documents |
| [`client.sources.extract()`](/sdk/extract) | Extract structured data using JSON Schema |
| [`client.sources.retrieve_chunks()`](/sdk/prebuilt-rag) | Retrieve relevant chunks for custom RAG |
| Method | Description |
| ------------------------------------------------------ | ----------------------------------------- |
| [`client.sources.ask()`](/sdk/chat) | Ask questions about your documents |
| [`client.sources.extract()`](/sdk/extract) | Extract structured data using JSON Schema |
| [`client.sources.retrieveChunks()`](/sdk/prebuilt-rag) | Retrieve relevant chunks for custom RAG |
## Complete Workflow Example
Here's the full "happy path": **ingest → get\_build\_status (poll) → list → get\_elements → chat/extract/retrieve\_chunks**; optionally **reprocess** by `file_id`.
### 1. Ingest a source
```python theme={null}
from pathlib import Path
from graphor import Graphor
import time
client = Graphor()
# Ingest returns build_id; poll get_build_status until ready
build_id = client.sources.ingest_file(file=Path("./document.pdf"))
while True:
status = client.sources.get_build_status(build_id)
if status.success and getattr(status, "file_id", None):
file_id = status.file_id
print(f"Ready. file_id: {file_id}")
break
time.sleep(2)
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
const client = new Graphor();
const buildId = await client.sources.ingestFile({ file: fs.createReadStream('./document.pdf') });
let fileId: string;
while (true) {
const status = await client.sources.getBuildStatus(buildId);
if (status.success && status.fileId) {
fileId = status.fileId;
console.log('Ready. file_id:', fileId);
break;
}
await new Promise((r) => setTimeout(r, 2000));
}
```
### 2. Reprocess (optional)
```python theme={null}
build_id = client.sources.reprocess(file_id=file_id, method="balanced")
print(f"Reprocessing: {build_id}")
```
```typescript theme={null}
const reprocessBuildId = await client.sources.reprocess({ file_id: fileId, method: 'balanced' });
console.log('Reprocessing:', reprocessBuildId);
```
### 3. List sources
```python theme={null}
sources = client.sources.list()
for s in sources:
print(f"{s.file_id} {s.file_name}: {s.status}")
target = next((s for s in sources if s.file_id == file_id), None)
```
```typescript theme={null}
const sources = await client.sources.list();
for (const s of sources) console.log(`${s.fileId} ${s.fileName}: ${s.status}`);
const target = sources.find((s) => s.fileId === fileId);
```
### 4. Get elements (after processing)
```python theme={null}
elements = client.sources.get_elements(file_id=file_id, page=1, page_size=50)
print(f"Total elements: {elements.total}")
for item in elements.items:
print(f" [{item.element_type}] {item.text[:100]}...")
```
```typescript theme={null}
const elements = await client.sources.getElements({ file_id: fileId, page: 1, page_size: 50 });
console.log(`Total elements: ${elements.total}`);
for (const item of elements.items) {
console.log(` [${item.element_type}] ${item.text.slice(0, 100)}...`);
}
```
### 5. Ask questions (Chat)
```python theme={null}
response = client.sources.ask(
question="What are the main topics in this document?",
file_ids=[file_id] # Optional: scope to specific sources
)
print(response.answer)
follow_up = client.sources.ask(
question="Can you elaborate on the first topic?",
conversation_id=response.conversation_id
)
print(follow_up.answer)
```
```typescript theme={null}
const response = await client.sources.ask({
question: 'What are the main topics in this document?',
fileIds: [fileId],
});
console.log(response.answer);
const followUp = await client.sources.ask({
question: 'Can you elaborate on the first topic?',
conversationId: response.conversation_id,
});
console.log(followUp.answer);
```
### 6. Extract data
```python theme={null}
result = client.sources.extract(
file_ids=[file_id],
user_instruction="Extract the invoice number and total amount.",
output_schema={
"type": "object",
"properties": {
"invoice_number": {"type": "string"},
"total_amount": {"type": "number"}
},
"required": ["invoice_number", "total_amount"]
}
)
print(result.structured_output)
```
```typescript theme={null}
const result = await client.sources.extract({
fileIds: [fileId],
user_instruction: 'Extract the invoice number and total amount.',
output_schema: {
type: 'object',
properties: {
invoice_number: { type: 'string' },
total_amount: { type: 'number' },
},
required: ['invoice_number', 'total_amount'],
},
});
console.log(result.structured_output);
```
### 7. Retrieve chunks (semantic search)
```python theme={null}
chunks = client.sources.retrieve_chunks(
query="What are the payment terms?",
file_ids=[file_id]
)
for chunk in chunks.chunks:
print(f"[{chunk.file_id}, Page {chunk.page_number}]", chunk.text[:80])
```
```typescript theme={null}
const chunks = await client.sources.retrieveChunks({
query: 'What are the payment terms?',
fileIds: [fileId],
});
for (const chunk of chunks.chunks) {
console.log(`[${chunk.fileId}, Page ${chunk.page_number}]`, chunk.text.slice(0, 80));
}
```
## Integration Patterns
### Complete SDK Client Wrapper
```python theme={null}
from graphor import Graphor, AsyncGraphor
import graphor
from pathlib import Path
from typing import Any
import os
class GraphorSDK:
"""Complete wrapper for common Graphor operations."""
def __init__(self, api_key: str | None = None):
self.client = Graphor(api_key=api_key) if api_key else Graphor()
# ==================== Sources ====================
def ingest_file(self, file_path: str | Path) -> str:
"""Ingest a file; returns build_id. Poll get_build_status for file_id."""
return self.client.sources.ingest_file(file=Path(file_path))
def get_build_status(self, build_id: str) -> Any:
"""Poll build status; when success, response has file_id."""
return self.client.sources.get_build_status(build_id)
def ingest_url(self, url: str, crawl: bool = False) -> str:
"""Ingest from a URL; returns build_id."""
return self.client.sources.ingest_url(url=url, crawl_urls=crawl)
def reprocess(self, file_id: str, method: str = "balanced") -> str:
"""Reprocess a source; returns build_id."""
return self.client.sources.reprocess(file_id=file_id, method=method)
def list_sources(self) -> list[dict[str, Any]]:
"""List all sources."""
sources = self.client.sources.list()
return [
{"file_id": s.file_id, "file_name": s.file_name, "status": s.status}
for s in sources
]
def get_elements(
self, file_id: str, page: int = 1, page_size: int = 50
) -> dict[str, Any]:
"""Get parsed elements from a source."""
result = self.client.sources.get_elements(
file_id=file_id, page=page, page_size=page_size
)
return {
"total": result.total,
"page": result.page,
"total_pages": result.total_pages,
"items": [
{"type": item.element_type, "content": item.text, "page": item.page_number}
for item in result.items
]
}
def delete(self, file_id: str) -> dict[str, Any]:
"""Delete a source by file_id."""
result = self.client.sources.delete(file_id=file_id)
return {"message": result.message}
# ==================== Chat ====================
def ask(
self,
question: str,
file_ids: list[str] | None = None,
conversation_id: str | None = None
) -> dict[str, Any]:
"""Ask a question about documents."""
kwargs = {"question": question}
if file_ids:
kwargs["file_ids"] = file_ids
if conversation_id:
kwargs["conversation_id"] = conversation_id
response = self.client.sources.ask(**kwargs)
return {"answer": response.answer, "conversation_id": response.conversation_id}
# ==================== Extraction ====================
def extract(
self, file_ids: list[str], instruction: str, schema: dict[str, Any]
) -> dict[str, Any]:
"""Extract structured data from documents."""
result = self.client.sources.extract(
file_ids=file_ids,
user_instruction=instruction,
output_schema=schema
)
return {"data": result.structured_output, "raw": result.raw_json}
# ==================== Retrieval ====================
def retrieve(
self, query: str, file_ids: list[str] | None = None
) -> dict[str, Any]:
"""Retrieve relevant chunks via semantic search."""
kwargs = {"query": query}
if file_ids:
kwargs["file_ids"] = file_ids
result = self.client.sources.retrieve_chunks(**kwargs)
return {
"query": result.query,
"total": result.total,
"chunks": [
{"text": c.text, "file_id": c.file_id, "page": c.page_number, "score": c.score}
for c in result.chunks or []
]
}
# Usage example
sdk = GraphorSDK()
import time
def full_workflow(file_path: str):
"""Complete ingestion, chat, and extraction workflow."""
try:
build_id = sdk.ingest_file(file_path)
while True:
status = sdk.get_build_status(build_id)
if status.success and getattr(status, "file_id", None):
file_id = status.file_id
break
time.sleep(2)
print(f"Ready: {file_id}")
chat_result = sdk.ask("Summarize this document", [file_id])
print(f"Summary: {chat_result['answer']}")
extract_result = sdk.extract(
[file_id],
"Extract key information",
{"type": "object", "properties": {"title": {"type": "string"}, "summary": {"type": "string"}}}
)
print(f"Extracted: {extract_result['data']}")
return {"success": True, "file_id": file_id}
except graphor.APIStatusError as e:
print(f"Error: {e}")
return {"success": False, "error": str(e)}
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
class GraphorSDK {
private client: Graphor;
constructor(apiKey?: string) {
this.client = apiKey ? new Graphor({ apiKey }) : new Graphor();
}
// ==================== Sources ====================
async ingestFile(filePath: string) {
return this.client.sources.ingestFile({ file: fs.createReadStream(filePath) });
}
async getBuildStatus(buildId: string) {
return this.client.sources.getBuildStatus(buildId);
}
async ingestUrl(url: string, crawl = false) {
return this.client.sources.ingestUrl({ url, crawlUrls: crawl });
}
async reprocess(fileId: string, method = 'balanced') {
return this.client.sources.reprocess({ file_id: fileId, method });
}
async listSources() {
const sources = await this.client.sources.list();
return sources.map((s) => ({ fileId: s.file_id, fileName: s.file_name, status: s.status }));
}
async getElements(fileId: string, page = 1, page_size = 50) {
const result = await this.client.sources.getElements({ file_id: fileId, page, page_size });
return {
total: result.total,
page: result.page,
totalPages: result.total_pages,
items: result.items.map((item) => ({ type: item.element_type, content: item.text, page: item.page_number })),
};
}
async delete(fileId: string) {
const result = await this.client.sources.delete({ file_id: fileId });
return { message: result.message };
}
// ==================== Chat ====================
async ask(question: string, fileIds?: string[], conversationId?: string) {
const response = await this.client.sources.ask({
question,
fileIds,
conversationId,
});
return { answer: response.answer, conversationId: response.conversation_id };
}
// ==================== Extraction ====================
async extract(fileIds: string[], instruction: string, schema: Record) {
const result = await this.client.sources.extract({
fileIds,
user_instruction: instruction,
output_schema: schema,
});
return { data: result.structured_output, raw: result.raw_json };
}
// ==================== Retrieval ====================
async retrieve(query: string, fileIds?: string[]) {
const result = await this.client.sources.retrieveChunks({ query, fileIds });
return {
query: result.query,
total: result.total,
chunks: (result.chunks ?? []).map((c) => ({ text: c.text, fileId: c.file_id, page: c.page_number, score: c.score })),
};
}
}
// Usage example
const sdk = new GraphorSDK();
async function fullWorkflow(filePath: string) {
try {
const buildId = await sdk.ingestFile(filePath);
let fileId: string;
while (true) {
const status = await sdk.getBuildStatus(buildId);
if (status.success && status.fileId) {
fileId = status.fileId;
break;
}
await new Promise((r) => setTimeout(r, 2000));
}
console.log(`Ready: ${fileId}`);
const chatResult = await sdk.ask('Summarize this document', [fileId]);
console.log(`Summary: ${chatResult.answer}`);
const extractResult = await sdk.extract(
[fileId],
'Extract key information',
{ type: 'object', properties: { title: { type: 'string' }, summary: { type: 'string' } } },
);
console.log(`Extracted: ${JSON.stringify(extractResult.data)}`);
return { success: true, fileId };
} catch (err) {
if (err instanceof Graphor.APIError) {
console.log(`Error: ${err.message}`);
return { success: false, error: err.message };
}
throw err;
}
}
```
### Async Integration
```python theme={null}
import asyncio
from graphor import AsyncGraphor
import graphor
class AsyncGraphorSDK:
"""Async wrapper for Graphor operations."""
def __init__(self, api_key: str | None = None):
self.client = AsyncGraphor(api_key=api_key) if api_key else AsyncGraphor()
async def process_multiple(
self, file_paths: list[str], method: str = "balanced"
) -> list[dict]:
"""Ingest multiple files concurrently; poll for file_id when needed."""
from pathlib import Path
import time
async def ingest_one(file_path: str) -> dict:
try:
build_id = await self.client.sources.ingest_file(file=Path(file_path))
while True:
status = await self.client.sources.get_build_status(build_id)
if status.success and getattr(status, "file_id", None):
return {"file": file_path, "status": "success", "file_id": status.file_id}
await asyncio.sleep(2)
except graphor.APIStatusError as e:
return {"file": file_path, "status": "failed", "error": str(e)}
return await asyncio.gather(*[ingest_one(fp) for fp in file_paths])
async def batch_ask(
self, questions: list[str], file_ids: list[str] | None = None
) -> list[dict]:
"""Ask multiple questions concurrently."""
async def ask_one(question: str) -> dict:
response = await self.client.sources.ask(
question=question, file_ids=file_ids
)
return {"question": question, "answer": response.answer}
return await asyncio.gather(*[ask_one(q) for q in questions])
# Usage
async def main():
sdk = AsyncGraphorSDK()
# Process multiple files
results = await sdk.process_multiple([
"doc1.pdf",
"doc2.pdf",
"doc3.pdf"
])
for r in results:
status = "OK" if r["status"] == "success" else "FAIL"
print(f"{status} {r['file']}")
# Ask multiple questions
answers = await sdk.batch_ask([
"What is the main topic?",
"Who are the key people mentioned?",
"What are the conclusions?"
])
for a in answers:
print(f"Q: {a['question']}")
print(f"A: {a['answer']}\n")
asyncio.run(main())
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
class AsyncGraphorSDK {
private client: Graphor;
constructor(apiKey?: string) {
this.client = apiKey ? new Graphor({ apiKey }) : new Graphor();
}
async processMultiple(filePaths: string[], method = 'balanced') {
const ingestOne = async (filePath: string) => {
try {
const buildId = await this.client.sources.ingestFile({ file: fs.createReadStream(filePath) });
let fileId: string;
while (true) {
const status = await this.client.sources.getBuildStatus(buildId);
if (status.success && status.fileId) {
fileId = status.fileId;
break;
}
await new Promise((r) => setTimeout(r, 2000));
}
return { file: filePath, status: 'success' as const, fileId };
} catch (err) {
const message = err instanceof Graphor.APIError ? err.message : String(err);
return { file: filePath, status: 'failed' as const, error: message };
}
};
return Promise.all(filePaths.map(ingestOne));
}
async batchAsk(questions: string[], fileIds?: string[]) {
const askOne = async (question: string) => {
const response = await this.client.sources.ask({ question, fileIds });
return { question, answer: response.answer };
};
return Promise.all(questions.map(askOne));
}
}
// Usage
async function main() {
const sdk = new AsyncGraphorSDK();
// Process multiple files
const results = await sdk.processMultiple(['doc1.pdf', 'doc2.pdf', 'doc3.pdf']);
for (const r of results) {
const status = r.status === 'success' ? 'OK' : 'FAIL';
console.log(`${status} ${r.file}`);
}
// Ask multiple questions
const answers = await sdk.batchAsk([
'What is the main topic?',
'Who are the key people mentioned?',
'What are the conclusions?',
]);
for (const a of answers) {
console.log(`Q: ${a.question}`);
console.log(`A: ${a.answer}\n`);
}
}
main();
```
## Error Handling
The SDK provides typed exceptions for different error scenarios:
```python theme={null}
import graphor
from graphor import Graphor
client = Graphor()
try:
build_id = client.sources.ingest_file(file=b"raw file contents")
except graphor.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__)
except graphor.RateLimitError as e:
print("Rate limit exceeded. Back off and retry.")
except graphor.BadRequestError as e:
print(f"Invalid request: {e}")
except graphor.AuthenticationError as e:
print(f"Invalid API key: {e}")
except graphor.NotFoundError as e:
print(f"Resource not found: {e}")
except graphor.APIStatusError as e:
print(f"API error (status {e.status_code}): {e}")
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
const client = new Graphor();
try {
const buildId = await client.sources.ingestFile({ file: fs.createReadStream('file.pdf') });
} catch (err) {
if (err instanceof Graphor.APIConnectionError) {
console.log('The server could not be reached');
console.log(err.cause);
} else if (err instanceof Graphor.RateLimitError) {
console.log('Rate limit exceeded. Back off and retry.');
} else if (err instanceof Graphor.BadRequestError) {
console.log(`Invalid request: ${err.message}`);
} else if (err instanceof Graphor.AuthenticationError) {
console.log(`Invalid API key: ${err.message}`);
} else if (err instanceof Graphor.NotFoundError) {
console.log(`Resource not found: ${err.message}`);
} else if (err instanceof Graphor.APIError) {
console.log(`API error (status ${err.status}): ${err.message}`);
} else {
throw err;
}
}
```
### Error Types
| Status Code | Error Type | Description |
| ----------- | -------------------------- | --------------------------------------- |
| 400 | `BadRequestError` | Invalid parameters or malformed request |
| 401 | `AuthenticationError` | Invalid or missing API key |
| 403 | `PermissionDeniedError` | Access denied to resource |
| 404 | `NotFoundError` | Resource doesn't exist |
| 422 | `UnprocessableEntityError` | Validation error |
| 429 | `RateLimitError` | Too many requests |
| ≥500 | `InternalServerError` | Server-side error |
| N/A | `APIConnectionError` | Network connectivity issues |
| N/A | `APITimeoutError` | Request timed out |
## Configuration
### Retries
Certain errors are automatically retried 2 times by default with exponential backoff:
```python theme={null}
from graphor import Graphor
# Configure default retries
client = Graphor(max_retries=0) # Disable retries
# Or per-request
client.with_options(max_retries=5).sources.ingest_file(file=b"...")
```
```typescript theme={null}
import Graphor from 'graphor';
// Configure default retries
const client = new Graphor({ maxRetries: 0 }); // Disable retries
// Or per-request
await client.sources.ask({ question: '...' }, { maxRetries: 5 });
```
### Timeouts
By default, requests time out after 1 minute:
```python theme={null}
from graphor import Graphor
# Configure default timeout (in seconds)
client = Graphor(timeout=120.0) # 2 minutes
# Or per-request
client.with_options(timeout=300.0).sources.reprocess(
file_id="file_abc123",
method="agentic"
)
```
By default, requests time out after 60 seconds:
```typescript theme={null}
import Graphor from 'graphor';
// Configure default timeout (in milliseconds)
const client = new Graphor({ timeout: 120 * 1000 }); // 2 minutes
// Or per-request
await client.sources.reprocess(
{ file_id: 'file_abc123', method: 'agentic' },
{ timeout: 300 * 1000 },
);
```
### Using aiohttp for Better Concurrency (Python only)
For high-concurrency async operations in Python, use the aiohttp client:
```python theme={null}
import asyncio
from graphor import AsyncGraphor, DefaultAioHttpClient
async def main():
async with AsyncGraphor(
http_client=DefaultAioHttpClient()
) as client:
# Your async operations here
sources = await client.sources.list()
print(f"Found {len(sources)} sources")
# Install aiohttp first: pip install graphor[aiohttp]
asyncio.run(main())
```
## Rate Limits and Best Practices
### Performance Guidelines
* **Batch Operations**: Process multiple files sequentially or with controlled concurrency
* **Async Processing**: Use `AsyncGraphor` (Python) or `Promise.all` (TypeScript) for concurrent operations
* **Retry Logic**: The SDK handles retries automatically; configure `max_retries` / `maxRetries` as needed
* **Timeout Handling**: Increase timeouts for large documents or complex processing
### Best Practices
```python theme={null}
from graphor import Graphor
import graphor
import time
client = Graphor(
max_retries=3,
timeout=120.0 # 2 minutes for processing operations
)
def robust_upload(file_path: str, max_attempts: int = 3) -> dict | None:
"""Ingest with custom retry logic; poll until ready and return file_id."""
from pathlib import Path
import time
for attempt in range(max_attempts):
try:
build_id = client.sources.ingest_file(file=Path(file_path))
while True:
status = client.sources.get_build_status(build_id)
if status.success and getattr(status, "file_id", None):
return {"success": True, "file_id": status.file_id}
time.sleep(2)
except graphor.RateLimitError:
wait_time = 2 ** attempt # Exponential backoff
print(f"Rate limited. Waiting {wait_time}s...")
time.sleep(wait_time)
except graphor.APIConnectionError as e:
print(f"Connection error (attempt {attempt + 1}): {e}")
time.sleep(1)
except graphor.APIStatusError as e:
print(f"API error: {e}")
return {"success": False, "error": str(e)}
return {"success": False, "error": "Max retries exceeded"}
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
const client = new Graphor({
maxRetries: 3,
timeout: 120 * 1000, // 2 minutes for processing operations
});
async function robustUpload(filePath: string, maxAttempts = 3) {
for (let attempt = 0; attempt < maxAttempts; attempt++) {
try {
const buildId = await client.sources.ingestFile({ file: fs.createReadStream(filePath) });
while (true) {
const status = await client.sources.getBuildStatus(buildId);
if (status.success && status.fileId) return { success: true, fileId: status.fileId };
await new Promise((r) => setTimeout(r, 2000));
}
} catch (err) {
if (err instanceof Graphor.RateLimitError) {
const waitTime = 2 ** attempt;
console.log(`Rate limited. Waiting ${waitTime}s...`);
await new Promise((r) => setTimeout(r, waitTime * 1000));
} else if (err instanceof Graphor.APIConnectionError) {
console.log(`Connection error (attempt ${attempt + 1}): ${err.message}`);
await new Promise((r) => setTimeout(r, 1000));
} else if (err instanceof Graphor.APIError) {
console.log(`API error: ${err.message}`);
return { success: false, error: err.message };
} else {
throw err;
}
}
}
return { success: false, error: 'Max retries exceeded' };
}
```
## Common Use Cases
### Document Processing Pipeline
```python theme={null}
from graphor import Graphor
from pathlib import Path
import graphor
client = Graphor(timeout=300.0)
import time
def document_pipeline(directory: str, partition_method: str = "balanced"):
"""Ingest all PDFs in a directory; poll until ready."""
results = []
for file_path in Path(directory).glob("*.pdf"):
try:
build_id = client.sources.ingest_file(file=file_path)
while True:
status = client.sources.get_build_status(build_id)
if status.success and getattr(status, "file_id", None):
results.append({"file": str(file_path), "status": "success", "file_id": status.file_id})
print(f"Ready: {status.file_id}")
break
time.sleep(2)
except graphor.APIStatusError as e:
results.append({"file": str(file_path), "status": "failed", "error": str(e)})
print(f"\nProcessed {sum(1 for r in results if r['status'] == 'success')}/{len(results)} files")
return results
```
```typescript theme={null}
import Graphor from 'graphor';
import fs from 'fs';
import path from 'path';
const client = new Graphor({ timeout: 300 * 1000 });
async function documentPipeline(directory: string, partitionMethod = 'balanced') {
const files = fs.readdirSync(directory).filter((f) => f.endsWith('.pdf'));
const results = [];
for (const file of files) {
const filePath = path.join(directory, file);
try {
const buildId = await client.sources.ingestFile({ file: fs.createReadStream(filePath) });
let fileId: string;
while (true) {
const status = await client.sources.getBuildStatus(buildId);
if (status.success && status.fileId) {
fileId = status.fileId;
console.log(`Ready: ${fileId}`);
break;
}
await new Promise((r) => setTimeout(r, 2000));
}
results.push({ file: filePath, status: 'success', fileId });
} catch (err) {
const message = err instanceof Graphor.APIError ? err.message : String(err);
results.push({ file: filePath, status: 'failed', error: message });
}
}
const successful = results.filter((r) => r.status === 'success').length;
console.log(`\nProcessed ${successful}/${results.length} files`);
return results;
}
```
### Q\&A System
```python theme={null}
from graphor import Graphor
client = Graphor()
class DocumentQA:
"""Simple Q&A system with conversation history."""
def __init__(self, file_ids: list[str] | None = None):
self.file_ids = file_ids
self.conversation_id = None
def ask(self, question: str) -> str:
"""Ask a question, maintaining conversation history."""
response = client.sources.ask(
question=question,
file_ids=self.file_ids,
conversation_id=self.conversation_id
)
# Store conversation ID for follow-up questions
self.conversation_id = response.conversation_id
return response.answer
def reset(self):
"""Reset conversation history."""
self.conversation_id = None
# Usage (file_ids from list())
qa = DocumentQA(file_ids=["file_abc123"])
print(qa.ask("What is this document about?"))
print(qa.ask("What are the main findings?")) # Follow-up
print(qa.ask("Can you summarize the conclusions?")) # Follow-up
qa.reset() # Start new conversation
```
```typescript theme={null}
import Graphor from 'graphor';
const client = new Graphor();
class DocumentQA {
private fileIds?: string[];
private conversationId?: string;
constructor(fileIds?: string[]) {
this.fileIds = fileIds;
}
async ask(question: string): Promise {
const response = await client.sources.ask({
question,
fileIds: this.fileIds,
conversationId: this.conversationId,
});
// Store conversation ID for follow-up questions
this.conversationId = response.conversation_id ?? undefined;
return response.answer;
}
reset() {
this.conversationId = undefined;
}
}
// Usage (fileIds from list())
const qa = new DocumentQA(['file_abc123']);
console.log(await qa.ask('What is this document about?'));
console.log(await qa.ask('What are the main findings?')); // Follow-up
console.log(await qa.ask('Can you summarize the conclusions?')); // Follow-up
qa.reset(); // Start new conversation
```
## Support and Resources
### Getting Help
Direct support for technical questions and issues
Learn how to generate and manage authentication tokens
Best practices for document upload and processing
Full REST API documentation for advanced use cases
## Next Steps
Ready to start building with the Graphor SDK? Choose your path:
### For Beginners
Ingest documents from files, URLs, GitHub, and YouTube; poll for file\_id
Ask natural language questions about your documents
Set up authentication for API access
### For Advanced Users
Extract structured data using JSON Schema
Retrieve relevant chunks via semantic search
Reprocess sources with different partition methods
Access structured document elements and metadata
The Graphor SDKs provide a powerful foundation for building intelligent, document-driven applications. With comprehensive support for document ingestion, conversational AI, structured extraction, and semantic search, both the Python and TypeScript SDKs give you the flexibility to build sophisticated AI workflows that scale from simple document search to complex analysis systems.