Metalogue
Docs

Python SDK

Full reference for the Metalogue Python SDK.

Python SDK

Official Python SDK for the Metalogue API.

Installation

pip install metalogue
# or
poetry add metalogue
# or
uv add metalogue

Quick Start

import os
from metalogue import MetalogueClient

client = MetalogueClient(api_key=os.environ["METALOGUE_API_KEY"])

# Federated query
response = await client.query("What was the auth decision?")

for result in response.results:
    print(f"[{result.source}] {result.content}")

Configuration

from metalogue import MetalogueClient, create_client

# Standard initialization
client = MetalogueClient(
    api_key="mlo_xxxxxxxxxxxx",
    base_url="https://api.metalogue.xyz",  # Optional
    timeout=30.0,                          # Optional, seconds
)

# Factory function (uses METALOGUE_API_KEY env var)
client = create_client()

Methods

Query

response = await client.query(
    "What was the decision on auth architecture?",
    limit=10,
    include_reasoning=True,
    sources=["slack", "notion"],
    date_after="2026-01-01",
    date_before="2026-01-31",
    authors=["alice@example.com"],
    types=["message", "document"],
)

# Response types
@dataclass
class QueryResponse:
    results: List[QueryResult]
    reasoning: Optional[str]
    query_id: str
    latency_ms: float

@dataclass
class QueryResult:
    id: str
    content: str
    source: str
    source_type: str
    score: float
    metadata: Dict[str, Any]
    url: Optional[str]

Connectors

# List all connectors
connectors = await client.list_connectors()

# Get specific connector
connector = await client.get_connector("slack-123")

# Create connector
new_connector = await client.create_connector(
    connector_type="slack",
    display_name="Company Slack",
    credentials={"token": "xoxb-..."},
    settings={"channels": ["general"]},
)

# Sync connector
await client.sync_connector("slack-123", full=True)

# Execute agentic action
await client.execute_action(
    "slack-123",
    intent="Send message to #general",
    context={"priority": "high"},
    dry_run=False,
)

# Delete connector
await client.delete_connector("slack-123")

Audit Logs

# Get audit logs
logs = await client.get_audit_logs(
    limit=100,
    action="query.executed",
    start="2026-01-01T00:00:00Z",
    end="2026-01-31T23:59:59Z",
)

# Export SOC2 report
report = await client.export_soc2(
    start_date="2026-01-01",
    end_date="2026-01-31",
)

Usage & Billing

# Get usage metrics
usage = await client.get_usage(period="month")

# Get subscription
subscription = await client.get_subscription()

Users

# List users
users = await client.list_users()

# Invite user
await client.invite_user("bob@example.com", role="member")

Error Handling

from metalogue import (
    MetalogueError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
)
import asyncio

try:
    response = await client.query("...")
except RateLimitError as e:
    print(f"Retry after {e.retry_after}s")
    await asyncio.sleep(e.retry_after)
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Resource not found")
except MetalogueError as e:
    print(f"API error: {e.message}")

Async Usage

The SDK is async by default:

import asyncio
from metalogue import MetalogueClient

async def main():
    client = MetalogueClient(api_key="mlo_xxx")
    response = await client.query("...")
    print(response.results)

asyncio.run(main())

Type Hints

Full type hints with Pydantic models:

from metalogue import (
    QueryResult,
    QueryResponse,
    Connector,
    AuditLogEntry,
    UsageMetrics,
)

response: QueryResponse = await client.query("...")
results: List[QueryResult] = response.results

Next Steps