dewey

Python client for the Dewey API. No third-party dependencies — uses only the Python standard library. See the full API reference for details on all endpoints and types.

Installation

pip install dewey

Quick start

from dewey import DeweyClient

client = DeweyClient(api_key="dwy_live_...")

# Create a collection (project_id from your Dewey dashboard → Project Settings)
col = client.collections.create("My Docs", project_id="proj_...")

# Upload a document
from pathlib import Path
doc = client.documents.upload(col.id, Path("report.pdf"))

# Query
results = client.retrieval.query(col.id, "What is the refund policy?")
for r in results:
    print(r.score, r.chunk.content[:100])

# Research (SSE streaming)
for event in client.research.stream(col.id, "Summarise key findings"):
    if event.type == "chunk":
        print(event.content, end="", flush=True)
    elif event.type == "done":
        print("\nSources:", event.sources)

Constructor

DeweyClient(api_key: str, base_url: str = "https://api.meetdewey.com/v1")

Resources

client.collections

update() accepts: name, visibility, chunk_size, chunk_overlap, description, enable_summarization, enable_captioning, enable_reranking, enable_deduplication, llm_model, instructions. llm_model and instructions accept None to clear the field; omit them entirely to leave unchanged.

# Set research instructions for a collection
client.collections.update(
    collection_id,
    instructions="All figures are in USD unless stated otherwise.",
)

# Clear instructions
client.collections.update(collection_id, instructions=None)

# Get collection statistics
stats = client.collections.stats(collection_id)
print(f"{stats.docCount} docs, {stats.totalClaimsCount} claims")

client.documents

upload() accepts a pathlib.Path, bytes, or any binary file-like object.

upload_many() is the recommended approach for large datasets. Each file is uploaded directly to S3 (bypassing the API server), so there are no payload-size limits. Files that match an existing document's hash are deduplicated automatically.

from pathlib import Path

docs = client.documents.upload_many(
    collection_id,
    list(Path("./reports").glob("**/*.pdf")),
    concurrency=10,
    on_progress=lambda doc, n, total: print(f"{n}/{total} {doc.filename}"),
)

Pass UploadManyItem instances when you need a custom filename or content type:

from dewey.resources.documents import UploadManyItem
from io import BytesIO

items = [
    UploadManyItem(file=BytesIO(data), filename="custom-name.pdf", content_type="application/pdf"),
]
docs = client.documents.upload_many(collection_id, items)

client.sections

client.retrieval

client.research

depth options: "quick", "balanced" (default), "deep", "exhaustive".

client.agents

Invoke saved agents defined in the dashboard. org_id and project_id are UUIDs (not slugs); agent_slug is the human-readable slug shown in dashboard URLs (e.g. "qa-test").

# Buffered — returns once the run terminates
result = client.agents.invoke_sync(
    org_id, project_id, "qa-test",
    query="What changed in 2023?",
)
print(result.response)
for s in result.sources:
    print(f"- {s.filename} § {s.sectionTitle}")

# Streaming — yields tool calls, tokens, and the final done event
for event in client.agents.stream(
    org_id, project_id, "qa-test",
    query="What changed in 2023?",
):
    if event.type == "chunk":
        print(event.content, end="", flush=True)
    elif event.type == "done":
        print(f"\n{len(event.sources)} sources")

AgentRunEvent is a tagged union: AgentRunEventStarted, AgentRunEventToolCall, AgentRunEventToolResult, AgentRunEventChunk, AgentRunEventDone, AgentRunEventError, AgentRunEventWarning. Match on event.type to discriminate.

client.claims

map_stream() yields raw event dicts. Check event["type"]: "progress", "done" (with claims list), or "error".

from dewey.types import ClaimMapItem

for event in client.claims.map_stream(collection_id):
    if event["type"] == "done":
        claims = [ClaimMapItem.from_dict(c) for c in event["claims"]]
        for claim in claims:
            print(f"[{claim.importance}] {claim.text}")

# Per-document claims (fast, no SSE)
result = client.claims.list_by_document(document_id, min_importance=3)
for claim in result.claims:
    print(claim.text)

client.contradictions

# Trigger detection
run = client.contradictions.detect(collection_id)
print("Run ID:", run.runId)

# Later: poll status
status = client.contradictions.get_latest_run(collection_id)
print(status.status, status.contradictionsFound)

# List active contradictions and apply resolutions
result = client.contradictions.list(collection_id, status="active")
for c in result.items:
    print(c.severity, c.explanation)
    # Apply the suggested resolution
    client.contradictions.apply_instruction(collection_id, c.id)

client.duplicates

Fuzzy document deduplication. Identifies near-duplicate documents by measuring how much content they share, marks one member of each cluster as canonical, and excludes near-duplicates from retrieval and contradiction detection. Must be enabled per-collection with client.collections.update(id, enable_deduplication=True).

# Enable on a collection (one-time)
client.collections.update(collection_id, enable_deduplication=True)

# Trigger detection, then poll
run = client.duplicates.detect(collection_id)
status = client.duplicates.get_latest_run(collection_id)
print(status.status, status.duplicateGroupsCreated)

# Review groups
result = client.duplicates.list(collection_id)
for group in result.items:
    for m in group.members:
        if m.relationship == "near_duplicate":
            pct = round((m.coverageToCanonical or 0) * 100)
            print(f"{m.filename} covers {pct}% of canonical")

client.provider_keys

Error handling

from dewey import DeweyClient, DeweyError

client = DeweyClient(api_key="dwy_live_...")

try:
    client.collections.get("unknown-id")
except DeweyError as e:
    print(e.status, e.message)  # e.g. 404 "Collection not found"

Presigned upload flow

For single files or when you need manual control, use the low-level presigned URL flow. For bulk ingestion, prefer upload_many() which handles this automatically with concurrency.

import hashlib
import urllib.request
from pathlib import Path

data = Path("file.pdf").read_bytes()
content_hash = hashlib.sha256(data).hexdigest()

# 1. Request a presigned URL
resp = client.documents.request_upload_url(
    collection_id,
    filename="file.pdf",
    content_type="application/pdf",
    file_size_bytes=len(data),
    content_hash=content_hash,
)

# 2. PUT bytes directly to S3 (no auth header needed)
req = urllib.request.Request(resp.uploadUrl, data=data, method="PUT")
req.add_header("Content-Type", "application/pdf")
urllib.request.urlopen(req)

# 3. Confirm to trigger ingestion
doc = client.documents.confirm(collection_id, resp.documentId)