Python SDK

The official Python SDK provides typed sync and async clients, automatic pagination, disk caching, retry logic, and data export.

Installation

pip install monopigi-sdk

Optional dependencies for data export:

pip install monopigi-sdk[polars]   # Parquet export + DataFrame conversion

Authentication

Pass your API key directly or store it via the CLI:

from monopigi import MonopigiClient

# Option 1: Pass directly
client = MonopigiClient(token="mp_live_YOUR_KEY")

# Option 2: Use stored credentials (from CLI login)
client = MonopigiClient()

# Option 3: Environment variable
import os
client = MonopigiClient(token=os.environ["MONOPIGI_API_KEY"])

Store credentials via the CLI:

monopigi auth login mp_live_YOUR_KEY

Client Configuration

client = MonopigiClient(
    token="mp_live_...",           # API key
    base_url="https://api.monopigi.com",  # Default
    max_retries=3,                 # Auto-retry on 429 (rate limit)
    cache_ttl=300,                 # Disk cache TTL in seconds (None = disabled)
)

The client supports context manager usage:

with MonopigiClient() as client:
    results = client.search("hospital")
    # connection is properly closed after the block

Core Methods

List Sources

sources = client.sources()
for src in sources:
    print(f"{src.name}: {src.label} ({src.status})")

Search

Full-text search across all or a specific source (Pro tier required):

# Search all sources
results = client.search("hospital procurement", limit=20)
print(f"Found {results.total} results")
for doc in results.results:
    print(f"  {doc.title} ({doc.source})")

# Search a specific source
results = client.search("renewable energy", source="rae", limit=10)

Browse Documents

Query documents from a specific source:

docs = client.documents("ted", limit=50, since="2026-01-01")
print(f"{docs.total} TED documents since 2026")
for doc in docs.documents:
    print(doc.title)

Statistics

stats = client.stats()
print(f"Total documents: {stats.total_documents}")
for name, info in stats.sources.items():
    print(f"  {name}: {info.documents} docs, avg quality {info.avg_quality:.2f}")

Usage

usage = client.usage()
print(f"Tier: {usage.tier}")
print(f"Used: {usage.daily_used}/{usage.daily_quota}")
print(f"Remaining: {usage.daily_remaining}")

Tier Info

# Check current tier
print(client.tier)  # "pro"

# Check rate limit info
print(client.quota)  # QuotaInfo(limit=5000, remaining=4958, reset=...)

# Check if a feature is available
if client.has_feature("full_text"):
    results = client.search("query")

Reports

Create and manage due diligence reports (Pro tier required):

# Create a report by AFM (VAT number)
report = client.create_report(entity_identifier="123456789", identifier_type="afm")
print(f"Report {report['id']} status: {report['status']}")

# List all reports
reports = client.list_reports(limit=10)
for r in reports["items"]:
    print(f"  {r['id']} — {r['entity_identifier']} ({r['status']})")

# Poll until complete
import time
while True:
    report = client.get_report(report["id"])
    if report["status"] == "completed":
        break
    time.sleep(5)

# Access the report data
print(report["report_json"])

# Download PDF
pdf_bytes = client.get_report_pdf(report["id"])
with open("report.pdf", "wb") as f:
    f.write(pdf_bytes)

Alerts

Create and manage procurement alert profiles (Pro tier required):

# Create an alert profile
profile = client.create_alert_profile(
    name="Hospital procurement",
    filters={
        "keywords": ["hospital", "medical equipment"],
        "sources": ["ted", "kimdis"],
        "min_value": 50000,
    },
    channels=["email", "webhook"],
    webhook_url="https://example.com/hooks/monopigi",
    notify_email="alerts@example.com",
)
print(f"Profile {profile['id']}: {profile['name']}")

# List profiles
profiles = client.list_alert_profiles()
for p in profiles["items"]:
    print(f"  {p['name']} — active={p['is_active']}")

# Update a profile
client.update_alert_profile(profile["id"], name="Updated name", is_active=False)

# Deactivate a profile
client.delete_alert_profile(profile["id"])

# List recent alert deliveries
deliveries = client.list_alert_deliveries(limit=20)
for d in deliveries["items"]:
    print(f"  {d['document_source_id']} via {d['channel']} ({d['delivery_status']})")

Monitoring

Add entities to continuous compliance monitoring (Enterprise tier required):

# Add an entity to monitor
entity = client.add_monitored_entity(
    entity_identifier="123456789",
    identifier_type="afm",
    label="ACME SA",
)
print(f"Monitoring {entity['label']} ({entity['id']})")

# List monitored entities
entities = client.list_monitored_entities()
for e in entities["items"]:
    print(f"  {e['label']} — last checked: {e['last_checked_at']}")

# Stop monitoring
client.remove_monitored_entity(entity["id"])

# List events for all monitored entities
events = client.list_entity_events(event_type="contract_award", since="2026-03-01T00:00:00Z")
for ev in events["items"]:
    print(f"  [{ev['event_type']}] {ev['summary']} (detected: {ev['detected_at']})")

# Acknowledge an event
client.acknowledge_event(ev["id"])

# Trigger a health report for a monitored entity
result = client.entity_health_report(entity["id"])
print(f"Report {result['report_id']} is {result['status']}")

Models

List available LLM models (no authentication required):

models = client.models()
for m in models.models:
    default = " (default)" if m.default else ""
    print(f"  {m.id}{default}")

Source-Specific Clients

Each data source has a typed sub-client:

with MonopigiClient() as client:
    # TED procurement notices
    notices = client.ted.notices(limit=10, since="2026-01-01")

    # Diavgeia government decisions
    decisions = client.diavgeia.decisions(limit=10)

    # ELSTAT statistical datasets
    datasets = client.elstat.datasets(limit=10)

    # RAE energy permits
    permits = client.rae.permits(limit=10)

    # data.gov.gr open data
    data = client.data_gov_gr.datasets(limit=10)

    # MITOS public service organizations
    orgs = client.mitos.organizations(limit=10)

Auto-Pagination

Iterator methods automatically handle pagination:

# Iterate through all search results
for doc in client.search_iter("public health", page_size=100):
    print(doc.title)

# Iterate through all documents from a source
for doc in client.documents_iter("diavgeia", since="2025-01-01"):
    process(doc)

Data Export

Export documents to JSON, CSV, or Parquet files:

# Export to JSON
count = client.export("ted", "ted_data.json", format="json")
print(f"Exported {count} documents")

# Export to CSV with date filter
client.export("diavgeia", "decisions.csv", format="csv", since="2026-01-01")

# Export to Parquet (requires polars)
client.export("elstat", "stats.parquet", format="parquet", limit=5000)

DataFrame Conversion

Search and document responses can be converted to Polars DataFrames:

results = client.search("procurement")
df = results.to_df()  # requires polars
print(df.head())

Async Client

For async applications (FastAPI, async scripts):

import asyncio
from monopigi import AsyncMonopigiClient

async def main():
    async with AsyncMonopigiClient() as client:
        results = await client.search("energy regulation")
        for doc in results.results:
            print(doc.title)

        sources = await client.sources()
        stats = await client.stats()

asyncio.run(main())

Caching

Enable disk caching to avoid redundant API calls:

# Cache responses for 5 minutes
client = MonopigiClient(cache_ttl=300)

# First call hits the API
sources = client.sources()

# Second call returns cached result (no API call)
sources = client.sources()

Cache files are stored in ~/.monopigi/cache/ as SHA-256-keyed JSON files.

Error Handling

from monopigi import MonopigiClient, AuthError, RateLimitError, NotFoundError, TierError

with MonopigiClient() as client:
    try:
        results = client.search("test")
    except AuthError:
        print("Invalid or missing API key")
    except RateLimitError as e:
        print(f"Rate limited, resets at: {e.reset_at}")
    except TierError:
        print("Feature not available on your tier — upgrade to Pro")
    except NotFoundError:
        print("Resource not found")

Response Format

All SDK methods return plain Python dicts matching the JSON responses from the API. See the API Reference for the full response schemas.