How to Integrate Haystack for payments with Elasticsearch for production AI

Combining Haystack for payments with Elasticsearch gives you a clean pattern for payment-aware AI agents: Haystack handles the retrieval and orchestration layer, while Elasticsearch stores and searches transaction records, payment policies, disputes, and merchant metadata at scale. That unlocks use cases like payment support copilots, fraud triage assistants, and compliance-aware agent workflows that can answer from live operational data instead of stale docs.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in your VPC
• •Access to the Haystack for payments SDK/package used in your environment
• •API credentials for your payment provider or sandbox environment
• •A working Python virtual environment
• •Network access from your app to Elasticsearch and the payment service
• •Basic knowledge of:
  • •REST APIs
  • •document indexing in Elasticsearch
  • •Haystack pipelines or components

Integration Steps

1. •

   Install the Python packages

   Keep the dependencies explicit. For production systems, pin versions and separate app runtime dependencies from dev tooling.

   pip install haystack elasticsearch python-dotenv
   

   If your Haystack for payments package is distributed under a different name in your environment, install that exact SDK as well.

2. •

   Connect to Elasticsearch and create an index for payment records

   Use Elasticsearch as the source of truth for searchable payment events, dispute notes, refund status, and merchant context.

   from elasticsearch import Elasticsearch
   
   es = Elasticsearch(
       "http://localhost:9200",
       basic_auth=("elastic", "changeme"),
       request_timeout=30,
   )
   
   index_name = "payments_events"
   
   if not es.indices.exists(index=index_name):
       es.indices.create(
           index=index_name,
           mappings={
               "properties": {
                   "payment_id": {"type": "keyword"},
                   "customer_id": {"type": "keyword"},
                   "status": {"type": "keyword"},
                   "amount": {"type": "double"},
                   "currency": {"type": "keyword"},
                   "notes": {"type": "text"},
                   "created_at": {"type": "date"},
               }
           },
       )
   

   This mapping keeps exact filters fast and full-text fields searchable. That matters when an agent needs to find “all failed card payments for merchant X in the last 24 hours.”

3. •

   Pull payment data from Haystack for payments and index it into Elasticsearch

   The integration point is simple: fetch normalized payment objects from Haystack’s payment client, then bulk index them into Elasticsearch for retrieval.

   from datetime import datetime, timezone
   from elasticsearch.helpers import bulk
   
   # Example SDK usage; replace with your actual Haystack for payments client import.
   from haystack_payments import PaymentsClient
   
   payments = PaymentsClient(
       api_key="PAYMENTS_API_KEY",
       base_url="https://api.your-payments-service.com",
   )
   
   recent_payments = payments.list_payments(limit=100)
   
   actions = []
   for p in recent_payments:
       actions.append(
           {
               "_index": index_name,
               "_id": p["payment_id"],
               "_source": {
                   "payment_id": p["payment_id"],
                   "customer_id": p["customer_id"],
                   "status": p["status"],
                   "amount": float(p["amount"]),
                   "currency": p["currency"],
                   "notes": p.get("notes", ""),
                   "created_at": p.get("created_at", datetime.now(timezone.utc).isoformat()),
               },
           }
       )
   
   bulk(es, actions)
   

   In production, run this on a schedule or via webhook-driven ingestion. Don’t query the payment API on every user prompt if you can avoid it.

4. •

   Search Elasticsearch from your AI agent before calling payment actions

   Your agent should retrieve context first: recent failures, prior refunds, risk notes, or merchant-specific rules. Then it can decide whether to call a payment action or answer directly.

   def search_payment_context(query: str):
       resp = es.search(
           index=index_name,
           size=5,
           query={
               "bool": {
                   "should": [
                       {"match": {"notes": query}},
                       {"match": {"status": query}},
                       {"match_phrase_prefix": {"payment_id": query}},
                   ]
               }
           },
       )
       return [hit["_source"] for hit in resp["hits"]["hits"]]
   
   
   context = search_payment_context("failed refund chargeback")
   
   
   # Example Haystack-style orchestration hook.
   # Use this context inside your agent prompt/tool selection logic.
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   
   

print(context)

5. **Wire retrieval + action into a single production flow**

   The pattern is: retrieve with Elasticsearch, reason with the agent, execute with Haystack for payments only when needed.

   ```python
   def handle_payment_support_request(customer_query: str):
       docs = search_payment_context(customer_query)

       if any(d["status"] == "failed" for d in docs):
           # Example SDK usage; replace with your actual method names.
           result = payments.list_refunds(payment_id=docs[0]["payment_id"])
           return {
               "context": docs,
               "refunds_or_actions": result,
           }

       return {
           "context": docs,
           "refunds_or_actions": [],
       }

   
response = handle_payment_support_request("Why did my card payment fail?")
print(response)

Testing the Integration

Use a known record path end-to-end: create or ingest one payment event, search it back through Elasticsearch, then call a read-only payment API method to confirm the connector works.

test_doc = {
    "_index": index_name,
    "_id": "pay_test_001",
    "_source": {
        "payment_id": "pay_test_001",
        "customer_id": "cust_123",
        "status": "failed",
        "amount": 49.99,
        "currency": "USD",
        "notes": "Test card declined by issuer",
        "created_at": datetime.now(timezone.utc).isoformat(),
    },
}

es.index(index=index_name, id=test_doc["_id"], document=test_doc["_source"])
es.refresh(index=index_name)

result = es.search(
    index=index_name,
    query={"term": {"payment_id.keyword": "pay_test_001"}},
)

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

Expected output:

{'payment_id': 'pay_test_001', 'customer_id': 'cust_123', 'status': 'failed', 'amount': 49.99, 'currency': 'USD', 'notes': 'Test card declined by issuer', 'created_at': '2026-04-21T...Z'}

If that passes, your indexing path is working and your agent can rely on Elasticsearch as the retrieval layer.

Real-World Use Cases

• •

  Payment support copilot

  • •Search failed charges, refunds, and dispute notes in Elasticsearch.
  • •Use Haystack for payments to fetch live transaction status before answering an operator or customer.
• •

  Fraud triage assistant

  • •Retrieve suspicious patterns like repeated declines or chargeback clusters.
  • •Trigger downstream review actions only when retrieval signals match risk rules.
• •

  Merchant operations agent

  • •Answer questions like “why did payouts slow down?” using indexed payout events and policy notes.
  • •Call payment APIs only when the agent needs fresh status or remediation 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.