Python SDK

For SDK methods, models, and errors, browse the API reference.

Deciding which credential to use? Choose an integration. This quick start uses your personal API token; an application can instead use a service account or user tokens from a dedicated MarcoPolo namespace.

Overview

Marcopolo provides one governed interface to enterprise systems. A connection may represent a database, warehouse, business application, API, or file provider. The SDK shares one product behavior with Marcopolo's MCP tools and web app, so permissions, semantics, and audit are identical no matter how a request arrives.

Get started

Install

pip install marcopolo-sdk

Create an API token

1. Sign in to the Marcopolo web app and open Developer Tools (mcp.marcopolo.dev/app/developer).
2. Choose Create Token, optionally name it, and pick how long it stays valid (30–365 days; 90 by default).
3. Copy the token. It is shown only once.

API tokens are available to Enterprise organizations.

Connect

import os
from marcopolo import Marcopolo

async with Marcopolo(access_token=os.environ["MARCOPOLO_API_TOKEN"]) as client:
    connections = await client.connections.list()

The token acts as you: every call sees your connections, runs in your workspace, and is audited as you. Keep it in a secret store, never in source control, and revoke it from Developer Tools when it is no longer needed. For automation shared by application users, follow the service account guide.

Other application identities

An Enterprise application can hold a service account API token when its users may share one set of connection grants. If each user needs their own workspace, permissions, connections, or audit identity, use a dedicated MarcoPolo namespace to issue a short-lived token for each authenticated user. Both tokens are passed to Marcopolo(access_token=...); the rest of the SDK workflow below is the same.

Use connections

connections = await client.connections.list()
for connection in connections:
    print(connection.name, connection.display_name, connection.connection_type)
    print(connection.auth_method, connection.share_mode, connection.is_owner)

connection = await client.connections.get("salesforce-prod")
configuration = await client.connections.get_configuration(connection.name)   # secrets masked

A connection is returned only when the authenticated user may use it: their own, shared with them, or shared with their organization. To learn what each connection can actually run, ask the workspace:

workspace = await client.workspace.connections()
for entry in workspace.connections:
    print(entry.connection.name, entry.capabilities)   # e.g. ['query', 'test', 'describe']

Run workspace commands

Use the same persistent workspace shell exposed to Marcopolo agents. The call waits up to timeout seconds. If the command is still alive, it returns an execution id instead of cancelling the work.

import asyncio
import json

result = await client.workspace.shell("python reports/build.py", timeout=30)
if result.status == "running":
    while True:
        await asyncio.sleep(5)
        poll = await client.workspace.shell(f"execution status {result.execution_id}")
        record = json.loads(poll.stdout)   # the build's execution record
        if record["status"] != "running":
            break
    if record["status"] == "failed":
        print(record["failure"]["kind"], record["failure"]["message"])
    else:
        print(record["outcome"]["stdout_tail"])
elif result.status == "failed":
    print(result.failure.kind, result.failure.message)
else:
    print(result.stdout)

execution status is itself a command, so its result's stdout holds the polled execution's record as JSON: status, failure, and an outcome with exit_code, stdout_tail, and stderr_tail. A lost execution, whose supervisor ended before recording one, has a null outcome.

A workspace permits one background command at a time. A non-zero exit code returns a failed result; transport, authentication, validation, and unavailable-workspace failures raise SDK exceptions.

Let users connect providers

For OAuth providers such as HubSpot, Salesforce, and GitHub, your application can start a hosted setup session through the SDK. Open the returned authorization URL for the user, then poll until the connection is ready. Marcopolo handles the provider callback and token exchange; provider credentials and tokens never pass through your application. Users can also add connections directly in the web app.

1 — Discover what users can connect

oauth_types = await client.connection_types.list(auth_method="oauth")
for t in oauth_types:
    print(t.connection_type, t.display_name, t.category)

2 — Start a setup session

started = await client.connection_setup.start(
    connection_type="hubspot",
    display_name="Joe's HubSpot",
)
# started.authorization_url — open this for the user
# started.setup_session_id  — poll this until terminal

Pass your application's stable client_session_id if a retry should reuse an in-flight setup session. Types with optional provider scopes accept requested_scopes. Provider availability depends on your deployment; if provider credentials are not configured there, setup returns a 503.

3 — Send the user to the provider

Open started.authorization_url in the user's browser. When authorization completes, the provider returns the user to Marcopolo's completion page unless you supplied a registered return_url.

4 — Check the session

session = await client.connection_setup.get(started.setup_session_id)
if session.status == "ready":
    result = await client.connections.test(session.connection_name)
elif session.status == "failed":
    print(session.failure.code, session.failure.message)

Poll every few seconds. The session state from the server is authoritative. Sessions expire 15 minutes after their last change; missing, expired, and another user's sessions all read as not found. The session belongs to the user, so polling can continue with a fresh token for that same user.

When the session is ready, the connection appears in client.connections.list() and can be queried. The user who authorized it owns the connection.

Return users to your application

With a dedicated MarcoPolo namespace, pass a return_url to bring users back to your application after provider authorization:

started = await client.connection_setup.start(
    connection_type="hubspot",
    display_name="Joe's HubSpot",
    return_url="https://app.example.com/connections/oauth-return",
)

The return URL must be absolute HTTPS and its origin must be registered for your namespace. An unregistered origin is rejected before contacting the provider. The browser returns with setup_session_id and status query parameters; treat those as a prompt to call connection_setup.get(), not as proof of success. Google Drive does not support an embedded return_url; start its setup without one.

Create connections directly

Every connection type publishes its ways of connecting as typed setup_methods. A hosted_oauth method uses the browser flow above; a fields method lists exactly which inputs to send to connections.create(), each with required and secret flags. Fields also carry what a form needs to render them: choices for enum dropdowns, item_type/min_items for lists, a file spec for file-backed inputs, group_label to head a dotted group like ssh_tunnel.*, and advanced to fold optional tuning behind a disclosure — enough to build a setup screen functionally equivalent to Marcopolo's own for field-based methods. Each type's icon_path is a public logo URL path (join it onto your service base URL) for rendering provider tiles.

postgres = await client.connection_types.get("pg")
for method in postgres.setup_methods:
    print(method.method, method.kind)            # e.g. 'manual' 'fields'
    for f in method.fields:
        print(f.name, f.type, f.required, f.secret, f.label)
        if f.choices:                            # enum fields ship their options
            print("  ", [(c.value, c.label) for c in f.choices])

# Declare the method and pass its inputs as one mapping — the server
# validates against the method and keeps secret values write-only:
connection = await client.connections.create(
    connection_type="pg",
    display_name="Warehouse",
    setup_method="manual",             # always declared: adding methods never breaks callers
    fields={"host": "db.internal", "dbname": "analytics",
            "user": "agent", "password": os.environ["WAREHOUSE_PASSWORD"]},
)

outcome = await client.connections.test(connection.name)
print(outcome.status, outcome.message)     # a failed test is an outcome, not an exception

connection = await client.connections.update(
    connection.name,
    display_name="Analytics Warehouse",
    configuration_patch={"database": "analytics_v2"},
)
await client.connections.delete(connection.name)

A submission that doesn't match the declared method is rejected with an error naming the problem — unknown fields, missing required fields, values whose shape contradicts the published field (wrong type, or an array below min_items), or a method that belongs to the hosted OAuth flow. Creation takes exactly this one form — setup_method plus fields — in the SDK and on the wire alike.

A field carrying a file spec takes file content as its value. Build it with FileUpload; the server validates the extension and decoded size against the field's published spec, and never accepts a server-side path.

from marcopolo import FileUpload

connection = await client.connections.create(
    connection_type="sqlite",
    display_name="Local Database",
    setup_method="manual",
    fields={"dbpath": FileUpload.from_path("analytics.db")},
)
# or, where the field's file spec allows it:
#   fields={"dbpath": FileUpload.create_empty()}

Share connections

from marcopolo import ShareScope

await client.connections.share(connection.name, ShareScope.USERS, users=["[email protected]"])
await client.connections.share(connection.name, ShareScope.COMPANY)
sharing = await client.connections.get_sharing(connection.name)
print(sharing.shared_with_company, sharing.shared_with_users)
await client.connections.unshare(connection.name, ShareScope.COMPANY)

New connections are private to their owner. Sharing requires management permission on the connection.

Query data

One method runs governed queries against every connection type. Its source names exactly one kind: SQL-like text (query_text), a saved workspace query (saved_query), or a provider operation (provider_operation) for APIs that take structured requests.

# Query text — databases, warehouses, SOQL, LogQL, ...
operation = await client.operations.query(
    "warehouse", source={"kind": "query_text", "text": "SELECT 1 AS one"}
)

# A saved query authored in the user's workspace, with parameters
operation = await client.operations.query(
    "warehouse",
    source={"kind": "saved_query", "path": "connections/warehouse/queries/daily_arr.sql"},
    parameters={"as_of": "2026-08-27"},
)

# A provider operation — REST-shaped connectors such as HubSpot, GitHub, Jira
operation = await client.operations.query(
    "joes-hubspot",
    source={
        "kind": "provider_operation",
        "definition": {
            "endpoint": "/crm/v3/objects/contacts",
            "method": "GET",
            "params": {"properties": "email,firstname,lastname", "limit": "100"},
            "paginate": True,
        },
    },
)

A provider_operation source's definition is passed to the connector verbatim; its shape is provider-specific and documented in that connector's query guide (catalog.query_guide), including which methods and endpoints it accepts. parameters template-substitutes into either query text or a saved query. input_reference chains a prior result reference in as this operation's input.

Learn the source before authoring

Connectors publish their catalog and a provider-specific query guide. Feeding both to your model is how a chatbot authors correct queries against a source it has never seen.

guide  = await client.catalog.query_guide("joes-hubspot")
print(guide.markdown)                                          # provider syntax, for your model's prompt

databases = await client.catalog.databases("warehouse")
tables    = await client.catalog.tables("warehouse", database="public")
columns   = await client.catalog.columns("warehouse", database="public", table="orders")
# or one call that reads whichever depth you ask for:
columns   = await client.catalog.inspect("warehouse", database="public", table="orders")

References, inline rows, and paging

By default a result is a reference: a named server-side relation with its own column schema and row count. Ask for inline rows only when they are small and immediately needed, and page any reference later.

operation = await client.operations.query("warehouse", source={"kind": "query_text", "text": sql})
assert operation.status == "succeeded", operation.failure

reference = operation.result.reference
print(reference.name, reference.row_count, reference.fields)

offset = 0
while offset is not None:
    page = await client.operations.records(operation.id, limit=100, offset=offset)
    for row in page.rows:
        ...
    offset = page.next_offset

small = await client.operations.query(
    "warehouse", source={"kind": "query_text", "text": sql}, inline=True, inline_limit=50
)
rows = small.result.records.rows        # inline_limit: 1–5000, default 500

References live in the user's workspace result store and follow its lifecycle: they are working data for the conversation at hand, not durable storage. A dropped reference answers records() with ResourceGoneError — re-run the query rather than persisting operation ids across sessions.

Compose results across sources

References are addressable in the built-in DUCKDB result store, so two large results can be joined where the data already lives, with only the summary reaching your application.

pipeline = await client.operations.query("salesforce-prod", source={"kind": "query_text", "text": soql})
revenue  = await client.operations.query("warehouse", source={"kind": "query_text", "text": sql})

answer = await client.operations.query(
    "DUCKDB",
    source={
        "kind": "query_text",
        "text": f"""
            SELECT p.account_id, sum(p.amount) AS pipeline, sum(r.amount) AS revenue
            FROM {pipeline.result.reference.name} p
            JOIN {revenue.result.reference.name} r USING (account_id)
            GROUP BY 1 ORDER BY pipeline - revenue DESC LIMIT 20
        """,
    },
    inline=True,
)

Each reference carries its own fields, so a model can author the join against results it has never read.

Build an agent

client.toolkit exposes Marcopolo as a set of agent tools — the same capabilities Marcopolo's own agents use. Each tool is bound to an SDK resource method, and its argument and result schemas come from the public API contract, so a tool call returns the same typed result or raises the same error as the method it wraps.

With LangChain or LangGraph, install the extra and pass the tools straight to your agent:

pip install "marcopolo-sdk[langchain]" langgraph langchain-anthropic

import os

from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from marcopolo import Marcopolo
from marcopolo.toolkit.langchain import langchain_tools

SYSTEM_PROMPT = """\
You answer questions using the company's governed data connections.
Use workspace_shell for workspace commands and querying data. Begin by reading
/workspace/README.md, /workspace/RULES.md and
/workspace/workflows/query-and-analyze-data.md, then follow the workflow.
Return a short, direct answer.
"""

async with Marcopolo(access_token=os.environ["MARCOPOLO_API_TOKEN"]) as client:
    agent = create_react_agent(
        ChatAnthropic(model="claude-sonnet-5-5", max_tokens=4096),
        tools=langchain_tools(client.toolkit),
        prompt=SYSTEM_PROMPT,
    )
    state = await agent.ainvoke(
        {"messages": [("user", "Which of our customers filed support tickets about invoices?")]}
    )

The agent works the way Marcopolo's own agents do: it reads /workspace/README.md, /workspace/RULES.md, and the connection's guidance through workspace_shell, authors a named query file, and runs connection query <name> --file <path> --json. Results stay in DuckDB as references, so intermediate rows never enter the prompt; the agent joins them through the DUCKDB connection and asks for --include-results only for a bounded final answer.

Failed executions and recoverable request errors come back to the model as error tool messages it can act on. Authentication, SDK-compatibility, transport, and invalid-response errors raise to your application, so an expired or revoked token surfaces where you can replace it.

For other frameworks, client.toolkit.tools() returns each tool's name, description, and JSON Schemas from the catalog bundled with the SDK, and client.toolkit.execute(name, arguments) runs it.

Handle errors

from marcopolo import (
    MarcopoloError, AuthenticationError, PermissionDeniedError,
    NotFoundError, ValidationError, UpstreamTimeoutError,
)

try:
    operation = await client.operations.query("warehouse", source={"kind": "query_text", "text": sql})
except AuthenticationError:
    ...                                                    # token expired or revoked: create a new one
except UpstreamTimeoutError:
    ...                                                    # 300s execution ceiling; nothing ran to completion
except MarcopoloError as error:
    print(error.status_code, error.code, error.message)
    print(error.retryable, error.correlation_id)           # quote correlation_id to support
else:
    if operation.status == "failed":
        print(operation.failure.category, operation.failure.message)

Every exception carries the server's error code, message, retryability, and a correlation id that Marcopolo can trace end to end. Synchronous execution is bounded at 300 seconds server-side; the client's default timeout is 60 seconds, so pass Marcopolo(..., timeout=330) when you expect long-running queries. SDKVersionUnsupportedError means the installed package is below the service's compatibility floor; upgrade before retrying. The full hierarchy and status mapping are in the SDK reference.

Security and permissions

• The credential determines the acting identity and its scope; neither is trusted from request parameters.
• An API token acts as the person who created it, expires on the date chosen at creation, and can be revoked at any time from Developer Tools.
• Provider secrets, access tokens, and refresh tokens never appear in SDK responses.
• Unknown, expired, or another user's setup sessions all read as not found.
• New connections are private to the user who created them until explicitly shared.
• With a namespace, the key can only issue user tokens; user tokens live five minutes and cannot mint further credentials.
• Every governed query emits an audit event attributed to the acting user, regardless of which client surface ran it.