Handling payments with PCI compliance

1
from livekit.agents import Agent, function_tool, RunContext
2

3
class PaymentAgent(Agent):
4
    def __init__(self):
5
        super().__init__(
6
            instructions="""You are a helpful sales assistant. When the customer
7
            is ready to pay, use the send_payment_link tool. Never ask for or
8
            accept credit card numbers directly."""
9
        )
10

11
    @function_tool()
12
    async def send_payment_link(
13
        self,
14
        amount: float,
15
        description: str,
16
        customer_phone: str,
17
    ) -> str:
18
        """Send a secure payment link to the customer's phone."""
19

20
        # Generate payment link via your PCI-compliant provider
21
        # (Stripe, Square, PayPal, etc.)
22
        payment_link = await create_payment_session(
23
            amount=amount,
24
            description=description,
25
            metadata={"session_id": self.session.id}
26
        )
27

28
        # Send via SMS
29
        await send_sms(
30
            to=customer_phone,
31
            body=f"Complete your payment securely: {payment_link}"
32
        )
33

34
        return "I've sent a secure payment link to your phone. Let me know once you've completed the payment."

1
@function_tool()
2
    async def check_payment_status(self, context: RunContext, session_id: str) -> str:
3
        """Check if the customer has completed their payment."""
4

5
        status = await get_payment_status(session_id)
6

7
        if status == "completed":
8
            return "Payment received successfully."
9
        elif status == "pending":
10
            return "Payment is still pending."
11
        else:
12
            return f"Payment status: {status}"

1
from livekit import api
2
from livekit.agents import Agent, function_tool, RunContext, get_job_context
3
from livekit.rtc import ParticipantKind
4

5
class SalesAgent(Agent):
6
    def __init__(self):
7
        super().__init__(
8
            instructions="""You are a sales agent. When ready to collect payment,
9
            transfer the caller to our secure payment system. Never collect
10
            card details yourself."""
11
        )
12

13
    @function_tool()
14
    async def transfer_to_payment_ivr(
15
        self,
16
        context: RunContext,
17
        amount: float,
18
        order_id: str,
19
    ) -> str:
20
        """Transfer caller to secure payment IVR."""
21

22
        job_ctx = get_job_context()
23

24
        # Find the SIP participant
25
        sip_participant = None
26
        for participant in job_ctx.room.remote_participants.values():
27
            if participant.kind == ParticipantKind.PARTICIPANT_KIND_SIP:
28
                sip_participant = participant
29
                break
30

31
        if not sip_participant:
32
            return "Unable to transfer - no phone connection found."
33

34
        # Transfer using the SIP API
35
        # Note: Your SIP trunk must have call transfers enabled
36
        try:
37
            await job_ctx.api.sip.transfer_sip_participant(
38
                api.TransferSIPParticipantRequest(
39
                    room_name=job_ctx.room.name,
40
                    participant_identity=sip_participant.identity,
41
                    transfer_to="sip:payment-ivr@your-pci-provider.com",
42
                )
43
            )
44
        except Exception as e:
45
            return f"Transfer failed: {e}"
46

47
        return "Transferring you to our secure payment line now."

1
await session.start(
2
    agent=agent,
3
    room=ctx.room,
4
    record=False,  # Disables audio, transcripts, traces, and logs upload
5
)

1
instructions="""CRITICAL COMPLIANCE RULES:
2
1. NEVER ask for credit card numbers, CVV, or expiration dates
3
2. NEVER repeat back any numbers that sound like card numbers
4
3. If a customer tries to give you card details, immediately redirect:
5
   "I can't accept card details directly. Let me send you a secure link instead."
6
4. For payments, either send a secure payment link OR transfer to our payment line
7
5. NEVER collect card details while the caller is connected to you
8
"""

1
import re
2
from typing import AsyncIterable, Optional
3
from livekit import rtc
4
from livekit.agents import Agent, ModelSettings, stt
5

6
class PCICompliantAgent(Agent):
7
    CARD_PATTERN = r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'
8

9
    async def stt_node(
10
        self,
11
        audio: AsyncIterable[rtc.AudioFrame],
12
        model_settings: ModelSettings
13
    ) -> Optional[AsyncIterable[stt.SpeechEvent]]:
14
        """Filter potential card numbers from transcription before LLM."""
15

16
        async for event in super().stt_node(audio, model_settings):
17
            # Check if this is a final transcript event
18
            if isinstance(event, stt.SpeechEvent) and event.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
19
                for alt in event.alternatives:
20
                    if re.search(self.CARD_PATTERN, alt.text):
21
                        # Replace with placeholder
22
                        alt.text = re.sub(
23
                            self.CARD_PATTERN,
24
                            "[CARD NUMBER BLOCKED]",
25
                            alt.text
26
                        )
27
                        # Queue a redirect message
28
                        self.session.say(
29
                            "I noticed you're trying to share card details. "
30
                            "For your security, I can't accept those directly. "
31
                            "Let me send you a secure payment link instead."
32
                        )
33
            yield event

1
from livekit.agents import Agent, AgentSession, AgentServer, JobContext, function_tool, RunContext
2
from livekit.plugins import openai, deepgram, silero
3
import httpx
4

5
class SecurePaymentAgent(Agent):
6
    def __init__(self):
7
        super().__init__(
8
            instructions="""You are a friendly order assistant for Acme Corp.
9

10
            Your job:
11
            1. Help customers with their orders
12
            2. Confirm order details (items, quantities, shipping)
13
            3. Calculate totals and explain charges
14
            4. When ready to pay, send a secure payment link
15

16
            NEVER collect card details directly. Always use send_payment_link."""
17
        )
18
        self.order_total = 0
19
        self.order_items = []
20

21
    @function_tool()
22
    async def add_to_order(
23
        self,
24
        context: RunContext,
25
        item: str,
26
        quantity: int,
27
        price: float
28
    ) -> str:
29
        """Add an item to the customer's order."""
30
        self.order_items.append({
31
            "item": item,
32
            "quantity": quantity,
33
            "price": price
34
        })
35
        self.order_total += price * quantity
36
        return f"Added {quantity}x {item} at ${price:.2f} each. Current total: ${self.order_total:.2f}"
37

38
    @function_tool()
39
    async def send_payment_link(self, context: RunContext, customer_phone: str) -> str:
40
        """Send secure payment link when customer is ready to pay."""
41

42
        if self.order_total <= 0:
43
            return "No items in the order yet. Please add items first."
44

45
        # Create Stripe payment session
46
        async with httpx.AsyncClient() as client:
47
            response = await client.post(
48
                "https://api.stripe.com/v1/payment_links",
49
                auth=("sk_live_xxx", ""),
50
                data={
51
                    "line_items[0][price_data][currency]": "usd",
52
                    "line_items[0][price_data][product_data][name]": "Order",
53
                    "line_items[0][price_data][unit_amount]": int(self.order_total * 100),
54
                    "line_items[0][quantity]": 1,
55
                }
56
            )
57
            payment_link = response.json()["url"]
58

59
        # Send SMS (implement your SMS provider here)
60
        await send_sms(customer_phone, f"Complete your ${self.order_total:.2f} payment: {payment_link}")
61

62
        return f"I've sent a secure payment link for ${self.order_total:.2f} to {customer_phone}. The link is valid for 24 hours. Let me know once you've completed the payment!"
63

64

65
server = AgentServer()
66

67
@server.rtc_session(agent_name="my-agent")
68
async def entrypoint(ctx: JobContext):
69
    session = AgentSession(
70
        llm=openai.LLM(model="gpt-4o"),
71
        stt=deepgram.STT(),
72
        tts=openai.TTS(),
73
        vad=silero.VAD.load(),
74
    )
75

76
    await session.start(
77
        agent=SecurePaymentAgent(),
78
        room=ctx.room,
79
        record=False,  # Disable recording for payment flows
80
    )