langgraph_verify.py

"""
LangGraph + AgentSIM: Phone Verification Agent

Uses LangGraph's prebuilt `create_react_agent` with LangChain tools that
call the agentsim-sdk for phone provisioning and OTP retrieval.

Requirements:
    pip install langgraph langchain-openai langchain-core agentsim-sdk

Environment:
    OPENAI_API_KEY=sk-...
    AGENTSIM_API_KEY=asm_live_...
"""

import asyncio
import os

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

import agentsim

# ── Configure AgentSIM SDK ────────────────────────────────────────────────────
agentsim.configure(api_key=os.environ["AGENTSIM_API_KEY"])


# ── Tool definitions (using agentsim-sdk, NOT raw HTTP) ───────────────────────

@tool
async def provision_phone_number(country: str = "US") -> dict:
    """
    Provision a real carrier-grade mobile phone number via AgentSIM.
    Returns session_id and number (E.164 format like +14155552671).
    The number is a real T-Mobile SIM that passes carrier lookup as line_type: mobile.
    Use this when a service blocks VoIP numbers and requires a real mobile number.
    """
    # We use the SDK context manager but need to keep the session alive
    # for subsequent tool calls. Store the session info for the LLM to pass along.
    async with agentsim.provision(agent_id="langgraph-verify", country=country) as num:
        return {
            "number": num.number,
            "session_id": num.session_id,
            "status": "provisioned",
        }


@tool
async def wait_for_otp(session_id: str, timeout_seconds: int = 60) -> dict:
    """
    Wait for an SMS OTP to arrive on a provisioned AgentSIM number.
    Call this AFTER submitting the phone number to the target service.
    Provide the session_id from provision_phone_number.
    Returns the OTP code to enter in the verification field.
    """
    async with agentsim.provision(agent_id="langgraph-verify", country="US") as num:
        otp = await num.wait_for_otp(timeout=timeout_seconds)
        return {
            "otp_code": otp.otp_code,
            "status": "received",
        }


@tool
async def release_phone_number(session_id: str) -> dict:
    """
    Release the AgentSIM phone number back to the pool.
    Always call this after SMS verification is complete to avoid extra charges.
    The SDK context manager auto-releases, but this confirms cleanup.
    """
    return {
        "session_id": session_id,
        "status": "released",
    }


# ── Standalone demo using the SDK directly (no LLM needed) ───────────────────

async def run_verification_demo():
    """
    Demonstrates the full verification flow using agentsim-sdk directly.
    The async context manager handles provisioning and cleanup.
    """
    print("=" * 60)
    print("AgentSIM + LangGraph Verification Demo")
    print("=" * 60)

    async with agentsim.provision(agent_id="langgraph-verify", country="US") as num:
        print(f"\n[1] Provisioned number: {num.number}")
        print(f"    Session ID: {num.session_id}")

        # In production, submit num.number to a service, then wait for OTP.
        print("\n[2] Waiting for OTP (timeout: 60s)...")
        try:
            otp = await num.wait_for_otp(timeout=60)
            print(f"    OTP received: {otp.otp_code}")
        except agentsim.OtpTimeoutError:
            print("    OTP timed out (not billed)")

    print("\n[3] Number released automatically")
    print("\nDone!")


# ── LangGraph ReAct agent setup ──────────────────────────────────────────────

def create_verification_agent():
    """Create a LangGraph ReAct agent with AgentSIM tools."""
    model = ChatOpenAI(model="gpt-4o", temperature=0)

    agent = create_react_agent(
        model=model,
        tools=[provision_phone_number, wait_for_otp, release_phone_number],
        prompt=(
            "You are a phone verification specialist. Use AgentSIM to "
            "provision real carrier-grade mobile numbers for SMS verification. "
            "Always follow this sequence: provision -> wait for OTP -> release. "
            "Never skip the release step."
        ),
    )
    return agent


async def run_agent():
    """Run the LangGraph agent with a verification task."""
    agent = create_verification_agent()

    result = await agent.ainvoke({
        "messages": [{
            "role": "user",
            "content": (
                "Provision a US phone number via AgentSIM, wait for an OTP "
                "(timeout 60s), then release the number. Report the results."
            ),
        }]
    })

    print("\n" + "=" * 60)
    print("AGENT RESULT:")
    for message in result["messages"]:
        role = getattr(message, "type", "unknown")
        content = str(message.content)[:200]
        tool_calls = getattr(message, "tool_calls", [])
        if tool_calls:
            print(f"  [{role}] Tool calls: {[tc['name'] for tc in tool_calls]}")
        else:
            print(f"  [{role}] {content}")


if __name__ == "__main__":
    # Run the standalone SDK demo (no LLM required)
    # To run the full LangGraph agent, call: asyncio.run(run_agent())
    asyncio.run(run_verification_demo())