Polyvia Python SDK

Official Python SDK for the Polyvia AI platform.

from polyvia import Polyvia

client = Polyvia(api_key="poly_<key>")

# Ingest → wait → query
result = client.ingest.file("report.pdf", name="Q4 Report")
client.ingest.wait(result.task_id)
print(client.query("What are the key findings?").answer)

---

Table of Contents

• Installation
• Authentication
• REST API
  • Ingest
  • Query
  • Groups
  • Documents
  • Usage & Rate Limits
• MCP Server
  • Anthropic beta MCP client
  • OpenAI Responses API
  • OpenAI Agents SDK
  • Claude Desktop
• Agent Tools (programmatic)
  • OpenAI ChatCompletion
  • Anthropic Messages API
  • LangChain
• Async Client
• Error Handling
• Development

---

Installation

pip install polyvia

LangChain agent support:

pip install "polyvia[langchain]"

Requires Python 3.9+.

---

Authentication

Generate an API key at app.polyvia.ai → Settings → API. All keys start with poly_.

# Pass explicitly
client = Polyvia(api_key="poly_<key>")

# Or set the environment variable and omit the argument
# export POLYVIA_API_KEY=poly_<key>
client = Polyvia()

Workspace scoping. Each key is permanently bound to the workspace (personal or one organization) you were in when you minted it. The key sees only that workspace's documents, groups, and chats — switching the active workspace in the UI later doesn't change a key's scope. Mint separate keys for each workspace you need to script against.

---

REST API

Ingest

# Single file — returns immediately with a task_id to poll
result = client.ingest.file("report.pdf", name="Q4 Report", group_id="g_<id>")
# IngestResult(document_id='<id>', task_id='<id>', status='pending')

# Multiple files in one request
batch = client.ingest.batch(
    ["q3.pdf", "q4.pdf"],
    names=["Q3 Report", "Q4 Report"],
    group_id="g_<id>",
)

# Check status
status = client.ingest.status(result.task_id)
# IngestionStatus(status='parsing', ...)

# Block until done — raises IngestionError on failure, IngestionTimeout on timeout
done = client.ingest.wait(result.task_id, poll_interval=5, timeout=300)

Query

# All completed documents
answer = client.query("What risks are mentioned across all reports?")

# Single document (fastest)
answer = client.query("Summarise section 3.", document_id="doc_<id>")

# Scoped to a group
answer = client.query("Key findings?", group_id="g_<id>")

# Scoped to multiple groups
answer = client.query("Compare results.", group_ids=["g_<id>", "g_<id>"])

print(answer.answer)

Groups

# Create
group = client.groups.create("Finance")
group_id = group["group_id"]

# List
for g in client.groups.list():
    print(g.name, g.id, g.color)

# Delete all documents in a group, then the group itself
client.groups.delete(group_id, delete_documents=True)

# Or separately
client.groups.delete_documents(group_id)   # wipe documents, keep group
client.groups.delete(group_id)             # remove empty group

Documents

# List — filter by status and/or group
docs = client.documents.list(status="completed", group_id="g_<id>")
docs = client.documents.list(group_ids=["g_<id>", "g_<id>"])

# Get one
doc = client.documents.get("doc_<id>")

# Move to a different group / remove from group
client.documents.update("doc_<id>", group_id="g_other")
client.documents.update("doc_<id>", group_id=None)

# Delete
client.documents.delete("doc_<id>")

Usage & Rate Limits

usage = client.usage()
print(usage.usage.requests.period)    # requests this calendar month
print(usage.usage.requests.total)     # all-time
print(usage.usage.documents_stored)  # live document count

limits = client.rate_limits()
print(limits.limits["requests_per_minute"])
print(limits.current["remaining_this_minute"])
print(limits.resets_at.month)         # ISO timestamp of next monthly reset

---

MCP Server

Polyvia runs a hosted Model Context Protocol server at https://app.polyvia.ai/mcp. Connect your AI client once and it can ingest, search, and query documents without any manual tool-dispatch code.

client.mcp returns an MCPConfig object with a helper for every major client:

Method	Use with
to_anthropic_mcp_server()	ant.beta.messages.create(mcp_servers=[...])
to_openai_responses_tool()	oai.responses.create(tools=[...])
to_openai_mcp_server()	OpenAI Agents SDK MCPServerStreamableHTTP
to_claude_desktop_config()	~/.claude/claude_desktop_config.json

---

Anthropic beta MCP client

from anthropic import Anthropic
from polyvia import Polyvia

polyvia = Polyvia(api_key="poly_<key>")
ant     = Anthropic()

response = ant.beta.messages.create(
    model="claude-opus-4-5",
    max_tokens=1000,
    messages=[{"role": "user", "content": "What are my Q4 findings?"}],
    mcp_servers=[polyvia.mcp.to_anthropic_mcp_server()],
    betas=["mcp-client-2025-04-04"],
)
print(response.content[0].text)

to_anthropic_mcp_server() produces:

{
    "type": "url",
    "url": "https://app.polyvia.ai/mcp",
    "name": "polyvia",            # customise with name="my-docs"
    "headers": {"Authorization": "Bearer poly_<key>"},
}

---

OpenAI Responses API

from openai import OpenAI
from polyvia import Polyvia

polyvia = Polyvia(api_key="poly_<key>")
oai     = OpenAI()

response = oai.responses.create(
    model="gpt-4o",
    tools=[polyvia.mcp.to_openai_responses_tool()],
    input="What are my Q4 findings?",
)
print(response.output_text)

to_openai_responses_tool() produces:

{
    "type": "mcp",
    "server_label": "polyvia",        # customise with server_label="my-docs"
    "server_url": "https://app.polyvia.ai/mcp",
    "headers": {"Authorization": "Bearer poly_<key>"},
    "require_approval": "never",      # or "always" to confirm each call
}

---

OpenAI Agents SDK

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHTTP
from polyvia import Polyvia

polyvia = Polyvia(api_key="poly_<key>")
cfg = polyvia.mcp.to_openai_mcp_server()

server = MCPServerStreamableHTTP(url=cfg["url"], headers=cfg["headers"])
agent  = Agent(name="Research", mcp_servers=[server])
result = Runner.run_sync(agent, "What do my Q4 reports say about revenue?")
print(result.final_output)

---

Claude Desktop

# Print a snippet to copy-paste into ~/.claude/claude_desktop_config.json
client.mcp.print_claude_desktop_snippet()

Or wire it up programmatically:

import json, pathlib

cfg_path = pathlib.Path.home() / ".claude" / "claude_desktop_config.json"
config = json.loads(cfg_path.read_text()) if cfg_path.exists() else {}
config.setdefault("mcpServers", {})["polyvia"] = client.mcp.to_claude_desktop_config()
cfg_path.write_text(json.dumps(config, indent=2))
print("Restart Claude Desktop to activate.")

to_claude_desktop_config() produces:

{
  "type": "http",
  "url": "https://app.polyvia.ai/mcp",
  "headers": { "Authorization": "Bearer poly_<key>" }
}

---

Agent Tools (programmatic)

If you'd rather manage the tool-dispatch loop yourself — or your framework doesn't support remote MCP — use client.tools to get JSON-schema tool definitions and an executor that calls the REST API directly.

All 10 Polyvia tools are included: ingest, status, list/get/update/delete documents, list/create/delete groups, and query.

OpenAI ChatCompletion

import json
from openai import OpenAI
from polyvia import Polyvia

client = Polyvia(api_key="poly_<key>")
oai    = OpenAI()

tools, call = client.tools.openai()

response = oai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What are my Q4 findings?"}],
    tools=tools,
)

for tc in response.choices[0].message.tool_calls or []:
    result = call(tc.function.name, json.loads(tc.function.arguments))
    print(result)

Anthropic Messages API

import anthropic
from polyvia import Polyvia

client = Polyvia(api_key="poly_<key>")
ant    = anthropic.Anthropic()

tools, call = client.tools.anthropic()

response = ant.messages.create(
    model="claude-opus-4-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Summarise my Finance documents."}],
    tools=tools,
)

for block in response.content:
    if block.type == "tool_use":
        result = call(block.name, block.input)
        print(result)

LangChain

Requires pip install "polyvia[langchain]".

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from polyvia import Polyvia

client = Polyvia(api_key="poly_<key>")
tools  = client.tools.langchain()

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to a document workspace."),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent    = create_tool_calling_agent(ChatOpenAI(model="gpt-4o"), tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
executor.invoke({"input": "What risks are mentioned in my reports?"})

---

Async Client

Every method on AsyncPolyvia is a coroutine — same API surface as the sync client.

import asyncio
from polyvia import AsyncPolyvia

async def main():
    async with AsyncPolyvia(api_key="poly_<key>") as client:
        result = await client.ingest.file("report.pdf")
        await client.ingest.wait(result.task_id)
        answer = await client.query("Key findings?")
        print(answer.answer)

asyncio.run(main())

---

Error Handling

from polyvia import (
    AuthenticationError,  # 401 — bad or missing API key
    ForbiddenError,        # 403 — document belongs to another user
    NotFoundError,         # 404 — document, group, or task not found
    RateLimitError,        # 429 — too many requests
    IngestionError,        # task finished with status='failed'
    IngestionTimeout,      # ingest.wait() exceeded its timeout
)

try:
    done = client.ingest.wait(task_id, timeout=60)
except IngestionError as e:
    print(f"Parsing failed: {e.error}")
except IngestionTimeout:
    print("Timed out — document may still be processing")
except RateLimitError:
    print("Rate limit hit — back off and retry")
except NotFoundError:
    print("Document or task not found")
except AuthenticationError:
    print("Invalid API key")

---

Development

git clone https://github.com/polyvia-ai/polyvia-python
cd polyvia-python
pip install -e ".[dev]"
pytest

---

License

MIT