Python SDK

Looking for every resource, method, model, and error? Browse the complete API reference.

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

Authenticate your users

A partner application holds a Marcopolo-issued namespace key (mpk_…) on its backend and exchanges it for a short-lived token for each end user it has already authenticated. The token pins the user's identity, tenant, and namespace; your application never passes a tenant identifier with requests.

import os
from marcopolo import Marcopolo, MarcopoloNamespace

async with MarcopoloNamespace(api_key=os.environ["MARCOPOLO_NAMESPACE_KEY"]) as ns:
    token = await ns.issue_user_token("joe@acme.com")
    # token.namespace, token.tenant  — where this user's data lives
    # token.expires_in               — 300 seconds

async with Marcopolo(access_token=token.access_token) as client:
    connections = await client.connections.list()

Issue tokens with each user's work email: coworkers automatically share a company scope (connections, sharing, audit) while other companies stay isolated, and consumer domains (gmail.com, outlook.com, …) are refused. Users who sign in to Marcopolo directly (or hold a developer token) pass their own credential as access_token; the client never interprets the credential type.

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:

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

Let users connect providers from your app

This is the flow for a "connect your data" page inside your product. Your backend starts a hosted setup session; your frontend sends the user to the provider in a popup; Marcopolo completes the provider callback and token exchange; your app observes the session until the connection is ready. Provider client secrets, authorization codes, and tokens never pass through your application.

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",
    return_url="https://app.example.com/connections/oauth-return",
    client_session_id=chat_session_id,   # retry-safe: same id + user + type reuses the in-flight session
)
# started.authorization_url  — open this for the user
# started.setup_session_id   — poll this until terminal

Types with optional provider scopes accept requested_scopes. Which providers are enabled for your deployment is discoverable at runtime through connection_types.list(auth_method="oauth") — a provider without credentials configured on the deployment answers setup with a 503.

3 — Send the user to the provider

Open started.authorization_url in a popup. When authorization completes, the user's browser lands on your return_url with setup_session_id and status query parameters — treat that as a hint to poll, not as truth.

4 — Observe until terminal

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; session state from the server is authoritative. Sessions expire fifteen minutes after their last change and then read as not found; an expired or foreign session is indistinguishable from a missing one. Because user tokens live five minutes, a backend polling across a slow authorization exchanges its namespace key for a fresh token and keeps the same setup_session_id — the session belongs to the user, not the token.

The connection is usable the moment the session reads ready: it appears in connections.list(), is owned by the end user, and accepts queries with no further setup step.

Create connections directly

Every connection type publishes its ways of connecting as typed setup_methods. Each method says what a client does (kind): a hosted_oauth method starts the flow above and collects nothing; a fields method lists exactly which inputs it takes, 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. Each type's icon_path is a public logo URL path (join it onto your service base URL) for rendering provider tiles. No provider documentation needed.

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 is satisfied through the Marcopolo app's upload flow; create rejects it by name rather than accepting a server-side path.

Share connections

from marcopolo import ShareScope

await client.connections.share(connection.name, ShareScope.USERS, users=["analyst@acme.com"])
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. Supply exactly one source: SQL-like text, a saved workspace query, or a provider operation for APIs that take structured requests.

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

# A saved query authored in the user's workspace, with parameters
operation = await client.operations.query(
    "warehouse",
    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",
    provider_operation={
        "endpoint": "/crm/v3/objects/contacts",
        "method": "GET",
        "params": {"properties": "email,firstname,lastname", "limit": "100"},
        "paginate": True,
    },
)

The provider_operation dict 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")

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", query_text=sql)
assert operation.status == "succeeded", operation.failure

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

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

small = await client.operations.query("warehouse", query_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", query_text=soql)
revenue  = await client.operations.query("warehouse", query_text=sql)

answer = await client.operations.query(
    "DUCKDB",
    query_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.

Handle errors

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

async def run_query_for(user_email: str, sql: str):
    token = await ns.issue_user_token(user_email)
    async with Marcopolo(access_token=token.access_token) as client:
        return await client.operations.query("warehouse", query_text=sql)

try:
    operation = await run_query_for("joe@acme.com", sql)
except AuthenticationError:
    operation = await run_query_for("joe@acme.com", sql)   # token expired mid-flight: re-exchange once
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 user and their company; neither is ever trusted from request parameters.
• A namespace key can only issue user tokens; user tokens live five minutes and cannot mint further credentials.
• Provider secrets, access tokens, and refresh tokens never appear in any SDK response, in either direction of the OAuth flow.
• OAuth return origins are registered per namespace, and unknown, expired, or foreign setup sessions all read as not found.
• New connections are private to the end user who created them until explicitly shared.
• Every governed query emits an audit event attributed to the acting user, regardless of which client surface ran it.