How to Integrate Haystack for payments with Elasticsearch for AI agents

Combining Haystack for payments with Elasticsearch gives you a practical pattern for payment-aware AI agents: retrieve the right policy, invoice, or transaction context fast, then let the agent decide what to do next. In banking and insurance workflows, that means fewer hallucinations, faster lookup on payment events, and better grounding when an agent needs to explain, validate, or route a payment-related request.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in your environment
• •A Haystack for payments account, API key, and access to its Python SDK
• •pip installed
• •Network access from your app to both services
• •Environment variables set for secrets:
  • •ELASTICSEARCH_URL
  • •ELASTICSEARCH_API_KEY or username/password
  • •HAYSTACK_PAYMENTS_API_KEY

Install the dependencies:

pip install elasticsearch haystack-payments python-dotenv

Integration Steps

1. •

   Initialize both clients

   Keep the connection setup in one place. For production systems, wrap this in a factory so your agent runtime can reuse clients instead of reconnecting on every request.

import os
from dotenv import load_dotenv
from elasticsearch import Elasticsearch
from haystack_payments import HaystackPaymentsClient

load_dotenv()

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

payments = HaystackPaymentsClient(
    api_key=os.environ["HAYSTACK_PAYMENTS_API_KEY"]
)

2. •

   Create an Elasticsearch index for payment context

   Store payment metadata your agent needs to retrieve quickly: customer ID, transaction ID, status, amount, and a short summary. Use a dedicated index so retrieval stays clean and queryable.

index_name = "payment_context"

if not es.indices.exists(index=index_name):
    es.indices.create(
        index=index_name,
        mappings={
            "properties": {
                "transaction_id": {"type": "keyword"},
                "customer_id": {"type": "keyword"},
                "status": {"type": "keyword"},
                "amount": {"type": "double"},
                "currency": {"type": "keyword"},
                "summary": {"type": "text"},
            }
        },
    )

3. •

   Fetch payment data from Haystack for payments and index it

   Pull the authoritative payment record from Haystack for payments, then write a normalized document into Elasticsearch. The key pattern is: source of truth in Haystack, retrieval layer in Elasticsearch.

def sync_payment(transaction_id: str):
    payment = payments.get_payment(transaction_id=transaction_id)

    doc = {
        "transaction_id": payment["transaction_id"],
        "customer_id": payment["customer_id"],
        "status": payment["status"],
        "amount": float(payment["amount"]),
        "currency": payment["currency"],
        "summary": (
            f"Payment {payment['transaction_id']} for customer {payment['customer_id']} "
            f"is {payment['status']} at {payment['amount']} {payment['currency']}"
        ),
    }

    es.index(index=index_name, id=transaction_id, document=doc)
    es.indices.refresh(index=index_name)

sync_payment("txn_100045")

4. •

   Query Elasticsearch inside your agent workflow

   Your AI agent should first retrieve relevant payment context from Elasticsearch before deciding whether to call more expensive tools or ask follow-up questions.

def get_payment_context(customer_id: str):
    response = es.search(
        index=index_name,
        query={
            "bool": {
                "filter": [
                    {"term": {"customer_id": customer_id}}
                ]
            }
        },
        size=5,
    )

    return [hit["_source"] for hit in response["hits"]["hits"]]

context = get_payment_context("cust_7781")
for item in context:
    print(item["summary"])

5. •

   Use both systems together in an agent action

   This is the real integration point. The agent can look up indexed context in Elasticsearch, then call Haystack for payments only when it needs authoritative state changes or fresh transaction details.

def resolve_payment_issue(customer_id: str, transaction_id: str):
    hits = es.search(
        index=index_name,
        query={
            "bool": {
                "must": [
                    {"term": {"customer_id": customer_id}},
                    {"term": {"transaction_id": transaction_id}},
                ]
            }
        },
        size=1,
    )

    if hits["hits"]["hits"]:
        cached = hits["hits"]["hits"][0]["_source"]
    else:
        cached = None

    live_payment = payments.get_payment(transaction_id=transaction_id)

    return {
        "cached_context": cached,
        "live_status": live_payment["status"],
        "can_refund": live_payment["status"] == "captured",
    }

result = resolve_payment_issue("cust_7781", "txn_100045")
print(result)

Testing the Integration

Run a quick end-to-end check: fetch a live payment record from Haystack for payments, store it in Elasticsearch, then read it back.

test_txn = "txn_100045"

payments_data = payments.get_payment(transaction_id=test_txn)

es.index(
    index=index_name,
    id=test_txn,
    document={
        "transaction_id": payments_data["transaction_id"],
        "customer_id": payments_data["customer_id"],
        "status": payments_data["status"],
        "amount": float(payments_data["amount"]),
        "currency": payments_data["currency"],
        "summary": f"Synced {payments_data['transaction_id']}",
    },
)

es.indices.refresh(index=index_name)

resp = es.search(
    index=index_name,
    query={"term": {"transaction_id": test_txn}},
)

print(resp["hits"]["hits"][0]["_source"])

Expected output:

{
  'transaction_id': 'txn_100045',
  'customer_id': 'cust_7781',
  'status': 'captured',
  'amount': 1250.0,
  'currency': 'USD',
  'summary': 'Synced txn_100045'
}

Real-World Use Cases

• •

  Payment support agents

  • •Retrieve recent transactions from Elasticsearch.
  • •Call Haystack for payments only when the user asks for authoritative status or refund eligibility.
• •

  Dispute resolution workflows

  • •Index chargeback notes, settlement timelines, and transaction metadata.
  • •Let the agent answer “what happened?” with grounded evidence instead of raw API calls every time.
• •

  Finance ops copilots

  • •Search failed payments by customer segment, merchant ID, or status.
  • •Trigger follow-up actions based on live payment state from Haystack for payments.

The production pattern is simple: Elasticsearch handles fast retrieval and filtering; Haystack for payments handles source-of-truth payment operations. Keep writes synchronized, read from Elasticsearch first, and reserve live API calls for state that must be current.

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.