> ## Documentation Index
> Fetch the complete documentation index at: https://docs.moneyline.co/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Official Python SDK for the Moneyline API — sync and async clients built on httpx, with pydantic v2 models for every response.

## Installation

```bash theme={null}
pip install moneyline
```

Requires Python 3.10 or newer.

## Quick start

```python theme={null}
from moneyline import Moneyline

# api_key defaults to the MONEYLINE_API_KEY environment variable
client = Moneyline()

submission = client.submissions.create(
    external_id="acme-001",
    name="Acme Corp",
)

client.documents.upload(submission.id, file="bank-statement.pdf")

scorecard = client.analytics.scorecard(submission.id)
```

> **Keyword arguments are `snake_case`, matching the API body.** Response models
> expose `snake_case` attributes too, so `submission.external_id` works even
> though the wire format is `externalId` — pydantic handles the aliasing.

## Configuration

```python theme={null}
client = Moneyline(
    api_key="ml_live_xxxxx",              # or MONEYLINE_API_KEY
    base_url="https://api.moneyline.co",  # or MONEYLINE_BASE_URL
    timeout=60.0,
    max_retries=3,
)
```

The client raises immediately if no API key is found.

## Async client

`AsyncMoneyline` mirrors the sync surface method-for-method; `await` the calls and iterate pages with `async for`.

```python theme={null}
import asyncio
from moneyline import AsyncMoneyline

async def main():
    client = AsyncMoneyline()
    submission = await client.submissions.create(external_id="acme-001")
    scorecard = await client.analytics.scorecard(submission.id)

    async for txn in client.transactions.list(submission_id=submission.id):
        print(txn.amount)

asyncio.run(main())
```

Note that `list(...)` itself is not awaited — it returns an `AsyncPage` you iterate.

## Pagination

List methods return a page object that transparently fetches subsequent pages as you iterate.

```python theme={null}
# Stream every match.
for txn in client.transactions.list(submission_id=submission.id):
    print(txn.posted_at, txn.amount, txn.category)

# Or collect into a list.
all_txns = list(client.transactions.list(submission_id=submission.id))
```

Methods returning a plain `list[...]` (`list_linked`, `event_types`, `groups`, `export`) are not paginated.

## Resources

### Submissions

```python theme={null}
sub = client.submissions.create(
    external_id="acme-001",
    name="Acme Corp",
    metadata={"source": "broker-portal"},
    company_info={"ein": "12-3456789"},
    idempotency_key="acme-001-create",
)

fetched = client.submissions.retrieve(sub.id)
updated = client.submissions.update(sub.id, name="Acme Corp LLC")

for s in client.submissions.list(status="processed", limit=25):
    print(s.external_id)

client.submissions.delete(sub.id)

client.submissions.start_workflow(sub.id)
status = client.submissions.get_status(sub.id)
client.submissions.enrich(sub.id, priority="high")
dupes = client.submissions.find_duplicates(sub.id)

# Polls until processed / reviewed / failed. Timeouts are in seconds.
done = client.submissions.wait_for_processing(sub.id, timeout=300, poll_interval=5)
```

### Documents

```python theme={null}
# `file` accepts a path, a Path, or an open binary file object.
doc = client.documents.upload(
    sub.id,
    file="statement.pdf",
    document_class="bank_statement",  # optional; skips auto-classification
)

fetched = client.documents.retrieve(doc.id)

for d in client.documents.list(submission_id=sub.id, document_class="bank_statement"):
    print(d.parsing_status)

raw = client.documents.get_content(doc.id)            # bytes
parsed = client.documents.get_parsed_results(doc.id)  # dict

client.documents.reclassify(doc.id, document_class="tax_return")
client.documents.parse(doc.id)
parts = client.documents.split(doc.id)

xlsx = client.documents.export(sub.id, format="xlsx")  # bytes
client.documents.delete(doc.id)
```

### Transactions

```python theme={null}
for txn in client.transactions.list(submission_id=sub.id):
    print(txn.description, txn.amount)

# Enrich without persisting (max 500 per request).
enriched = client.transactions.enrich(
    [{"description": "AMZN Mktp US*2A4B", "amount": -42.11, "date": "2026-01-14"}],
)

# Persist against a submission — submission_id is positional.
client.transactions.create(sub.id, [{"description": "DEPOSIT", "amount": 5000}])

# Stream enrichment results as they complete.
for result in client.transactions.stream_enrichment([{"description": "SHELL OIL"}]):
    print(result)

client.transactions.delete(txn_id)
```

### Analytics

Every calculation lives here, keyed by submission ID.

```python theme={null}
scorecard  = client.analytics.scorecard(sub.id)
cashflow   = client.analytics.cashflow(sub.id)
risk       = client.analytics.risk_score(sub.id)
balances   = client.analytics.balances(sub.id)
anomalies  = client.analytics.anomalies(sub.id)
forecast   = client.analytics.forecast(sub.id)
statistics = client.analytics.statistics(sub.id)
bank       = client.analytics.bank_summary(sub.id)
recurring  = client.analytics.recurring(sub.id)
merchants  = client.analytics.merchants(sub.id)
categories = client.analytics.categories(sub.id)
metrics    = client.analytics.metrics(sub.id)
benchmarks = client.analytics.benchmarks(sub.id)
coverage   = client.analytics.coverage(sub.id)
financials = client.analytics.financials(sub.id)
```

### Policies

```python theme={null}
evaluations = client.policies.evaluate(sub.id)
summary = client.policies.evaluate_all(sub.id)
client.policies.progress(sub.id, action="approve", reason="Meets all criteria")

funder_policies = client.policies.list_funder_policies(funder_id)
```

### Rules

```python theme={null}
rule = client.rules.create(
    name="Low average balance",
    metric="average_daily_balance",
    operator="lt",
    threshold=5000,
)

fetched = client.rules.retrieve(rule.id)
client.rules.update(rule.id, threshold=7500)
for r in client.rules.list():
    print(r.name)
client.rules.delete(rule.id)

metrics = client.rules.available_metrics()
```

### Webhooks

```python theme={null}
hook = client.webhooks.create(
    url="https://example.com/moneyline",
    events=["submission.processed", "document.parsed"],
)

fetched = client.webhooks.retrieve(hook.id)
client.webhooks.update(hook.id, is_active=False)
for w in client.webhooks.list():
    print(w.url)
client.webhooks.delete(hook.id)

client.webhooks.test(hook.id)
types = client.webhooks.event_types()

for delivery in client.webhooks.deliveries(hook.id, limit=50):
    print(delivery.status_code)
client.webhooks.replay(hook.id, delivery_id)
```

### Merchants and categories

```python theme={null}
merchant = client.merchants.retrieve(merchant_id)
results = client.merchants.search(query="amazon")
groups = client.merchants.groups()
resolved = client.merchants.extract(description="AMZN Mktp US*2A4B9")

categories = client.categories.list()
```

### Funders

```python theme={null}
for f in client.funders.list():
    print(f.name)
funder = client.funders.retrieve(funder_id)
linked = client.funders.list_linked()

client.funders.submit(funder_id=funder_id, submission_id=sub.id)
submissions = client.funders.list_submissions(funder_id)
```

### Integrations

```python theme={null}
integration = client.integrations.create(type="plaid", name="Plaid production")

for i in client.integrations.list():
    print(i.type)
client.integrations.update(integration.id, is_active=False)
links = client.integrations.list_links(integration.id)

# Note: `import` is a Python keyword, so this method is `import_data`.
client.integrations.import_data(
    submission_id=sub.id,
    source="plaid",
    data={"access_token": "..."},
)
```

### Entities

```python theme={null}
entities = client.entities.list()
entity = client.entities.retrieve(entity_id)
client.entities.bulk_save(items=[{"id": entity_id, "name": "Beneficial owner"}])
exported = client.entities.export()
```

### ISO applications

```python theme={null}
app = client.iso_applications.upload(file_path="iso-application.pdf")
# or: client.iso_applications.upload(file_bytes=data, filename="app.pdf")

fetched = client.iso_applications.retrieve(app.id)
results = client.iso_applications.results(app.id)
client.iso_applications.reprocess(app.id)
```

## Error handling

```python theme={null}
from moneyline.errors import (
    MoneylineError,
    AuthenticationError,
    PermissionError,
    NotFoundError,
    ValidationError,
    ConflictError,
    RateLimitError,
    ServerError,
)

try:
    client.submissions.retrieve("does-not-exist")
except NotFoundError:
    print("no such submission")
except ValidationError as err:
    print("bad request:", err)
except RateLimitError:
    print("rate limited after retries")
except MoneylineError as err:
    print(err.code, err.request_id)
```

`RateLimitError` and `ServerError` are retried automatically up to `max_retries` before surfacing.

## Webhook signature verification

```python theme={null}
from moneyline.webhook import construct_event, WebhookSignatureVerificationError

@app.post("/moneyline")
async def handle(request):
    try:
        event = construct_event(
            await request.body(),            # raw bytes, not the parsed JSON
            request.headers["x-moneyline-signature"],
            os.environ["MONEYLINE_WEBHOOK_SECRET"],
        )
    except WebhookSignatureVerificationError:
        return Response(status_code=400)
    return Response(status_code=200)
```

`verify_signature` performs the same check without parsing the payload.

## Generated models

Spec-driven pydantic models are generated from the OpenAPI document (`scripts/gen-types.sh`) and re-exported by the entity models:

```python theme={null}
from moneyline.types.generated import Submission, Transaction
```

Use these when you need the exact wire shape of an endpoint the hand-written resources do not wrap.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.