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)- Installation
- Authentication
- REST API
- MCP Server
- Agent Tools (programmatic)
- Async Client
- Error Handling
- Development
pip install polyviaLangChain agent support:
pip install "polyvia[langchain]"Requires Python 3.9+.
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.
# Single file — pass the group **name** (created if it doesn't exist yet).
# Returns immediately with a task_id to poll.
result = client.ingest.file("report.pdf", name="Q4 Report", group="Finance")
# IngestResult(document_id='<id>', task_id='<id>', status='pending')
# Multiple files in one request (the group is resolved once for the batch)
batch = client.ingest.batch(
["q3.pdf", "q4.pdf"],
names=["Q3 Report", "Q4 Report"],
group="Finance",
)
# 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)# 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 — by name (the group must already exist)
answer = client.query("Key findings?", group="Finance")
# Scoped to multiple groups — by id
answer = client.query("Compare results.", group_ids=["g_<id>", "g_<id>"])
print(answer.answer)Groups have a human name and an opaque backend id. You rarely need the
id — pass the name straight to ingest / query (above) and the SDK resolves
it. When you do want the group object, get_or_create is the easy way in.
# Idempotent: returns the existing "Finance" group, or creates it. Matched by
# exact name, so it never makes duplicates. Returns a Group.
group = client.groups.get_or_create("Finance")
group.id # the backend id, if you ever need it
group.name # "Finance"
# Look one up without creating it (returns None if there isn't one)
existing = client.groups.find("Finance")
# 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
# create() always makes a NEW group, even if the name exists — prefer
# get_or_create() unless you specifically want a fresh one each time.# 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 = 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 resetPolyvia 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.
Using Claude Code? Add the server with one command:
claude mcp add --transport http polyvia https://app.polyvia.ai/mcp \
--header "Authorization: Bearer poly_<key>"client.mcp returns an MCPConfig object with a helper for every major client:
| Method | Use with |
|---|---|
claude_code_command() |
The claude mcp add … command line above |
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 |
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>"},
}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
}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)# 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>" }
}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.
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)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)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?"})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())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")git clone https://github.com/polyvia-ai/polyvia-python
cd polyvia-python
pip install -e ".[dev]"
pytestMIT