graphor 0.21.0

The official Python library for the graphor API

Graphor Python SDK

The official Python SDK for the Graphor API. Build intelligent document applications with ease.

Features:

• 📄 Document Ingestion — Upload files, web pages, GitHub repos, and YouTube videos
• 💬 Document Chat — Ask questions with conversational memory
• 📊 Structured Extraction — Extract data using JSON Schema
• 🔍 Semantic Search — Retrieve relevant chunks for custom RAG pipelines
• ⚡ Async Support — Full async/await support with AsyncGraphor
• 🔒 Type Safety — Complete type definitions for all params and responses

Documentation

📚 Full documentation: docs.graphorlm.com/sdk/overview

Installation

pip install graphor

For better async performance with aiohttp:

pip install graphor[aiohttp]

Quick Start

from graphor import Graphor

client = Graphor()  # Uses GRAPHOR_API_KEY env var

# Upload a document
source = client.sources.upload(file=open("document.pdf", "rb"))
print(f"Uploaded: {source.file_name}")

# Process with advanced parsing
client.sources.parse(
    file_name=source.file_name, 
    partition_method="graphorlm" # Options: basic (Fast), hi_res (Balanced), hi_res_ft (Accurate), mai (VLM), graphorlm (Agentic)
)

# Ask questions about your documents
response = client.sources.ask(question="What are the main topics?")
print(f"Answer: {response.answer}")

Authentication

Set your API key as an environment variable (recommended):

export GRAPHOR_API_KEY="grlm_your_api_key_here"

from graphor import Graphor

client = Graphor()  # Automatically uses GRAPHOR_API_KEY

Or pass it directly:

client = Graphor(api_key="grlm_your_api_key_here")

Core Features

📄 Upload Documents

Upload files, web pages, GitHub repositories, or YouTube videos:

from pathlib import Path
from graphor import Graphor

client = Graphor()

# Upload a local file
source = client.sources.upload(file=Path("report.pdf"))

# Upload from URL
source = client.sources.upload_url(url="https://example.com/article")

# Upload from GitHub
source = client.sources.upload_github(url="https://github.com/org/repo")

# Upload from YouTube
source = client.sources.upload_youtube(url="https://youtube.com/watch?v=...")

Supported formats: PDF, DOCX, TXT, MD, HTML, CSV, XLSX, PNG, JPG, MP3, MP4, and more.

📖 Full upload documentation

Certain errors are automatically retried 0 times by default, with a short exponential backoff. Connection errors (for example, due to a network connectivity problem), 408 Request Timeout, 409 Conflict, 429 Rate Limit, and >=500 Internal errors are all retried by default.

⚙️ Process Documents

Reprocess documents with different OCR/parsing methods:

# Reprocess with high-resolution parsing
source = client.sources.parse(
    file_name="document.pdf",
    partition_method="hi_res"  # Options: basic, hi_res, hi_res_ft, mai, graphorlm
)

📖 Full processing documentation

💬 Chat with Documents

Ask questions about your documents with conversational memory:

# Ask a question
response = client.sources.ask(
    question="What are the key findings?"
)
print(response.answer)

# Follow-up question (maintains context)
follow_up = client.sources.ask(
    question="Can you elaborate on the first point?",
    conversation_id=response.conversation_id
)
print(follow_up.answer)

# Scope to specific documents
response = client.sources.ask(
    question="Compare these two reports",
    file_names=["report-2023.pdf", "report-2024.pdf"]
)

📖 Full chat documentation

📊 Extract Structured Data

Extract structured information using JSON Schema:

result = client.sources.extract(
    file_names=["invoice.pdf"],
    user_instruction="Extract invoice details",
    output_schema={
        "type": "object",
        "properties": {
            "invoice_number": {"type": "string"},
            "total_amount": {"type": "number"},
            "line_items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "description": {"type": "string"},
                        "amount": {"type": "number"}
                    }
                }
            }
        }
    }
)

print(result.structured_output)
# {"invoice_number": "INV-001", "total_amount": 1500.00, "line_items": [...]}

📖 Full extraction documentation

🔍 Retrieve Chunks (Prebuilt RAG)

Build custom RAG pipelines with semantic search:

# Retrieve relevant chunks
result = client.sources.retrieve_chunks(
    query="What are the payment terms?"
)

for chunk in result.chunks:
    print(f"[{chunk.file_name}, Page {chunk.page_number}]")
    print(f"Score: {chunk.score:.2f}")
    print(chunk.text)
    print("---")

# Use with your preferred LLM
context = "\n".join([c.text for c in result.chunks])
# Pass context to OpenAI, Anthropic, etc.

📖 Full RAG documentation

📋 Manage Sources

List, inspect, and delete documents:

# List all sources
sources = client.sources.list()
for source in sources:
    print(f"{source.file_name}: {source.status}")

# Get document elements
elements = client.sources.load_elements(
    file_name="document.pdf",
    page=1,
    page_size=50
)

# Delete a source
result = client.sources.delete(file_name="document.pdf")

Async Usage

Use AsyncGraphor for async/await support:

import asyncio
from graphor import AsyncGraphor

async def main():
    client = AsyncGraphor()
    
    # All methods support await
    source = await client.sources.upload(file=b"content")
    response = await client.sources.ask(question="Summarize this document")
    
    print(response.answer)

asyncio.run(main())

With aiohttp (Better Concurrency)

from graphor import AsyncGraphor, DefaultAioHttpClient

async def main():
    async with AsyncGraphor(http_client=DefaultAioHttpClient()) as client:
        sources = await client.sources.list()
        print(f"Found {len(sources)} sources")

asyncio.run(main())

File uploads

Request parameters that correspond to file uploads can be passed as bytes, or a PathLike instance or a tuple of (filename, contents, media type).

from pathlib import Path
from graphor import Graphor

client = Graphor()

client.sources.ingest_file(
    file=Path("/path/to/file"),
)

The async client uses the exact same interface. If you pass a PathLike instance, the file contents will be read asynchronously automatically.

Error Handling

import graphor
from graphor import Graphor

client = Graphor()

try:
    response = client.sources.ask(question="What is this about?")
except graphor.AuthenticationError:
    print("Invalid API key")
except graphor.NotFoundError:
    print("Resource not found")
except graphor.RateLimitError:
    print("Rate limited - back off and retry")
except graphor.APIConnectionError:
    print("Network error")
except graphor.APIStatusError as e:
    print(f"API error: {e.status_code}")

Status Code	Error Type
400	BadRequestError
401	AuthenticationError
403	PermissionDeniedError
404	NotFoundError
422	UnprocessableEntityError
429	RateLimitError
≥500	InternalServerError
N/A	APIConnectionError

Configuration

Retries

Requests are automatically retried twice with exponential backoff:

# Configure default retries
client = Graphor(max_retries=5)

# Or per-request
client.with_options(max_retries=3).sources.ask(question="...")

Timeouts

Default timeout is 60 seconds:

# Configure default timeout
client = Graphor(timeout=120.0)

# Or per-request
client.with_options(timeout=300.0).sources.parse(
    file_name="large-document.pdf",
    partition_method="graphorlm"
)

Complete Example

from pathlib import Path
from graphor import Graphor

client = Graphor()

# 1. Upload a document
source = client.sources.upload(file=Path("contract.pdf"))
print(f"✅ Uploaded: {source.file_name}")

# 2. Process with advanced parsing
processed = client.sources.parse(
    file_name=source.file_name,
    partition_method="hi_res"
)
print(f"✅ Processed: {processed.status}")

# 3. Ask questions
response = client.sources.ask(
    question="What are the key terms of this contract?",
    file_names=[source.file_name]
)
print(f"📝 Answer: {response.answer}")

# 4. Extract structured data
extracted = client.sources.extract(
    file_names=[source.file_name],
    user_instruction="Extract contract details",
    output_schema={
        "type": "object",
        "properties": {
            "parties": {"type": "array", "items": {"type": "string"}},
            "effective_date": {"type": "string"},
            "termination_date": {"type": "string"},
            "total_value": {"type": "number"}
        }
    }
)
print(f"📊 Extracted: {extracted.structured_output}")

# 5. Build custom RAG
chunks = client.sources.retrieve_chunks(
    query="payment obligations",
    file_names=[source.file_name]
)
print(f"🔍 Found {chunks.total} relevant chunks")

MCP Server

Use the Graphor MCP Server to enable AI assistants to interact with this API, allowing them to explore endpoints, make test requests, and use documentation to help integrate this SDK into your application.

Or manually add it to your MCP client's configuration:

{
  "mcpServers": {
    "graphor_api": {
      "command": "npx",
      "args": ["-y", "graphor-mcp@latest"],
      "env": {
        "GRAPHOR_API_KEY": "grlm_your_api_key_here"
      }
    }
  }
}

Note: You may need to set environment variables in your MCP client.

Remote MCP Server (Web Apps & Agentic Workflows)

For web-based AI clients (e.g. Claude.ai) or agentic frameworks (e.g. LangChain, CrewAI) that cannot run local npx processes, use the hosted remote MCP server. Authentication is handled via OAuth — a browser window will open for you to log in.

https://mcp.graphor.workers.dev/sse

Web apps (e.g. Claude.ai) — in Claude.ai, go to Settings > Connectors > Add custom connector, fill in the name and the remote MCP server URL. You will be redirected to log in through the OAuth flow:

https://mcp.graphor.workers.dev/sse

Desktop clients (e.g. Claude Desktop) — use mcp-remote as a local proxy:

{
  "mcpServers": {
    "graphor_api": {
      "command": "npx",
      "args": ["mcp-remote", "https://mcp.graphor.workers.dev/sse"]
    }
  }
}

Agentic workflows (e.g. LangChain) — connect via SSE transport:

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
    {
        "graphor": {
            "url": "https://mcp.graphor.workers.dev/sse",
            "transport": "sse",
        }
    }
) as client:
    tools = client.get_tools()
    # Use tools with your LangChain agent

See the MCP Server README for more details.

API Reference

Sources

Method	Description	Docs
sources.upload()	Upload a local file	📖
sources.upload_url()	Upload from web URL	📖
sources.upload_github()	Upload from GitHub	📖
sources.upload_youtube()	Upload from YouTube	📖
sources.parse()	Reprocess with different method	📖
sources.list()	List all sources	📖
sources.delete()	Delete a source	📖
sources.load_elements()	Get parsed elements	📖

Chat & AI

Method	Description	Docs
sources.ask()	Ask questions about documents	📖
sources.extract()	Extract structured data	📖
sources.retrieve_chunks()	Retrieve chunks for RAG	📖

Requirements

• Python 3.9+

Contributing

See CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.

Links

• 📚 Documentation
• 🐛 Issue Tracker
• 📦 PyPI
• 🏠 Graphor