How to Integrate Haystack for payments with Elasticsearch for startups

Combining Haystack for payments with Elasticsearch gives you a clean pattern for payment-aware AI agents: one system handles payment orchestration, the other handles retrieval over invoices, transaction history, chargeback notes, and support cases. For startups, this is useful when you need an agent that can answer billing questions, reconcile payment events, or route disputes using both structured payment data and searchable operational context.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in the cloud
• •A Haystack for payments account, API key, and project/workspace configured
• •pip installed
• •Network access from your app to both services
• •Environment variables set:
  • •HAYSTACK_PAYMENTS_API_KEY
  • •HAYSTACK_PAYMENTS_BASE_URL
  • •ELASTICSEARCH_URL
  • •ELASTICSEARCH_API_KEY if using Elastic Cloud

Install the Python dependencies:

pip install elasticsearch haystack-ai requests

Integration Steps

1. •Set up your clients

Create dedicated clients for Elasticsearch and Haystack for payments. Keep them isolated so retries, auth, and timeouts are easy to manage.

import os
from elasticsearch import Elasticsearch
from haystack import Pipeline
from haystack.components.builders import PromptBuilder

es = Elasticsearch(
    os.environ["ELASTICSEARCH_URL"],
    api_key=os.environ.get("ELASTICSEARCH_API_KEY"),
)

# Example pattern for a payments SDK client.
# Replace with the actual client class from your Haystack for payments package.
from haystack_payments import Client as PaymentsClient

payments = PaymentsClient(
    api_key=os.environ["HAYSTACK_PAYMENTS_API_KEY"],
    base_url=os.environ["HAYSTACK_PAYMENTS_BASE_URL"],
)

2. •Index payment events into Elasticsearch

Store payment events as documents so your agent can search them later by customer ID, invoice ID, status, or error code.

payment_event = {
    "payment_id": "pay_123",
    "customer_id": "cus_456",
    "invoice_id": "inv_789",
    "status": "failed",
    "amount": 4900,
    "currency": "USD",
    "error_code": "card_declined",
    "created_at": "2026-04-21T10:15:00Z",
}

es.index(
    index="payment-events",
    id=payment_event["payment_id"],
    document=payment_event,
)

If you are also pulling events from Haystack for payments, normalize them before indexing.

events = payments.payments.list(limit=20)

for event in events.data:
    es.index(
        index="payment-events",
        id=event.id,
        document={
            "payment_id": event.id,
            "customer_id": event.customer_id,
            "invoice_id": event.invoice_id,
            "status": event.status,
            "amount": event.amount,
            "currency": event.currency,
            "error_code": getattr(event, "error_code", None),
            "created_at": event.created_at,
        },
    )

3. •Query Elasticsearch before calling payment actions

Use Elasticsearch as the retrieval layer. The agent should first search for context, then decide whether it needs to call a payment action.

query = {
    "query": {
        "bool": {
            "must": [
                {"term": {"customer_id.keyword": "cus_456"}},
                {"terms": {"status.keyword": ["failed", "pending"]}},
            ]
        }
    },
    "_source": ["payment_id", "invoice_id", "status", "error_code", "amount"],
}

result = es.search(index="payment-events", body=query)

for hit in result["hits"]["hits"]:
    print(hit["_source"])

4. •Trigger a payment workflow from Haystack for payments

Once the agent has enough context from Elasticsearch, call the relevant payment API method. A common startup pattern is retrying a failed charge or creating a support-friendly payment link.

failed_payment_id = result["hits"]["hits"][0]["_id"]

retry_response = payments.payments.retry(failed_payment_id)

print(retry_response.id)
print(retry_response.status)

If your use case is invoice collection instead of retries:

invoice_link = payments.invoices.create_payment_link(
    invoice_id="inv_789"
)

print(invoice_link.url)

5. •Wire retrieval and payment logic into a single agent flow

This is where the integration becomes useful. The agent retrieves context from Elasticsearch, then decides whether to call Haystack for payments.

def get_payment_context(customer_id: str):
    response = es.search(
        index="payment-events",
        body={
            "query": {
                "term": {"customer_id.keyword": customer_id}
            },
            "_source": ["payment_id", "status", "error_code", "invoice_id"],
            "sort": [{"created_at": {"order": "desc"}}],
            "size": 5,
        },
    )
    return [hit["_source"] for hit in response["hits"]["hits"]]


def retry_latest_failed_payment(customer_id: str):
    events = get_payment_context(customer_id)
    failed = next((e for e in events if e["status"] == "failed"), None)

    if not failed:
        return {"message": f"No failed payments found for {customer_id}"}

    retry_result = payments.payments.retry(failed["payment_id"])
    
    es.index(
        index="payment-events",
        id=f"{failed['payment_id']}-retry",
        document={
            **failed,
            "status_after_retry": retry_result.status,
        },
    )

    return retry_result


print(retry_latest_failed_payment("cus_456"))

Testing the Integration

Run a simple end-to-end test: write one failed payment event to Elasticsearch, query it back, and trigger the retry path.

test_doc = {
    "payment_id": "pay_test_001",
    "customer_id": "cus_test_001",
    "invoice_id": "inv_test_001",
    "status": "failed",
    "amount": 1200,
    "currency": "USD",
    "error_code": None,
}

es.index(index="payment-events", id=test_doc["payment_id"], document=test_doc)

search = es.search(
    index="payment-events",
    body={"query": {"term": {"customer_id.keyword": test_doc["customer_id"]}}},
)

print(search["hits"]["total"])
print(search["hits"]["hits"][0]["_source"]["status"])

Expected output:

{'value': 1, 'relation': 'eq'}
failed

If you have sandbox credentials for Haystack for payments, verify the retry call returns a valid status object:

pay_test_001 retried

Real-World Use Cases

• •Payment support agent: retrieve recent failures from Elasticsearch and let the agent decide whether to retry a charge or create a support ticket.
• •Dunning workflow assistant: search overdue invoices and generate follow-up actions through Haystack for payments.
• •Finance ops copilot: answer questions like “why did this customer fail twice?” by combining indexed event history with live payment actions.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.