from typing import TypedDict, List, Optional, Dict

class InsuranceState(TypedDict):
    """Simple state for product price insurance workflow."""
    # Input
    product_query: str

    # Node outputs
    search_results: Dict[str, str]  # platform -> URL
    price_data: List[Dict[str, any]]  # extracted prices
    final_report: str

    # Optional fields
    error: Optional[str]
    confidence_score: Optional[float]

from pydantic import BaseModel, Field
from typing import Optional

class ProductURLs(BaseModel):
    """Structured output for extracted product URLs."""
    amazon: Optional[str] = Field(None, description="Amazon product page URL")
    walmart: Optional[str] = Field(None, description="Walmart product page URL") 
    bestbuy: Optional[str] = Field(None, description="Best Buy product page URL")

async def search_products(state: InsuranceState) -> InsuranceState:
    product_query = state["product_query"]

    try:
        # Configure MCP client for Bright Data
        browserai_config = {
            "mcpServers": {
                "BrightData": {
                    "command": "npx",
                    "args": ["@brightdata/mcp"],
                    "env": {
                        "API_TOKEN": os.getenv("BRIGHT_DATA_API_TOKEN"),
                        "WEB_UNLOCKER_ZONE": os.getenv("WEB_UNLOCKER_ZONE", "unblocker")
                    }
                }
            }
        }

        client = MCPClient.from_dict(browserai_config)
        adapter = LangChainAdapter()
        tools = await adapter.create_tools(client)

        # Create ReAct agent with LLM + tools
        agent = create_react_agent(
            model=llm,
            tools=tools,
            prompt="Find direct product pages on Amazon, Walmart, and Best Buy..."
        )

        result = await agent.ainvoke({
            "messages": [{"role": "user", "content": f"Find product pages for: {product_query}"}]
        })

        # Use structured output instead of regex parsing
        structured_llm = llm.with_structured_output(ProductURLs)
        url_response = await structured_llm.ainvoke(
            f"Extract product URLs from these search results:\n\n{result['messages'][-1].content}"
        )

        # Filter out None values
        urls = {k: v for k, v in url_response.dict().items() if v is not None}

        return {**state, "search_results": urls}

    except Exception as e:
        return {**state, "search_results": {}, "error": f"Search failed: {str(e)}"}

class ExtractedPrice(BaseModel):
    """Structured output for extracted price information."""
    price: Optional[float] = Field(None, description="Product price as number")
    title: str = Field("", description="Product name/title")
    availability: str = Field("Unknown", description="Availability status")

async def extract_prices(state: InsuranceState) -> InsuranceState:
    search_results = state.get("search_results", {})

    if not search_results:
        return {**state, "price_data": [], "error": "No URLs found"}

    price_data = []

    # Configure MCP client and create extraction agent
    # ... (similar setup as search node)

    for platform, url in search_results.items():
        try:
            # Use platform-specific extractors
            extraction_prompt = f"Extract price, title, and availability from: {url}"
            result = await agent.ainvoke({"messages": [{"role": "user", "content": extraction_prompt}]})

            # Use structured output for reliable parsing
            structured_llm = llm.with_structured_output(ExtractedPrice)
            structured_data = await structured_llm.ainvoke(
                f"Extract price data for {platform}:\n\n{result['messages'][-1].content}"
            )

            price_data.append({
                "platform": platform,
                "price": structured_data.price,
                "title": structured_data.title,
                "url": url,
                "availability": structured_data.availability
            })

        except Exception as e:
            price_data.append({
                "platform": platform,
                "price": None,
                "title": "",
                "url": url,
                "availability": "Error extracting",
                "error": str(e)
            })

    return {**state, "price_data": price_data}

import statistics

async def generate_report(state: InsuranceState) -> InsuranceState:
    price_data = state.get("price_data", [])
    product_query = state["product_query"]

    # Extract valid prices
    valid_prices = [item["price"] for item in price_data if item.get("price") is not None]

    if not valid_prices:
        return {
            **state,
            "final_report": f"❌ No prices found for {product_query}",
            "confidence_score": 0.0
        }

    # Calculate statistics
    median_price = statistics.median(valid_prices)
    average_price = statistics.mean(valid_prices)
    min_price = min(valid_prices)
    max_price = max(valid_prices)

    # Calculate confidence score
    platforms_found = len(valid_prices)
    confidence_score = min(10.0, (platforms_found / 3.0) * 8.0 + 2.0)

    # Use LLM to format the report
    report_prompt = ChatPromptTemplate.from_messages([
        ("system", "Create a clean, professional price analysis report with emojis and clear sections..."),
        ("user", "Create report for: {product}\nMedian: ${median:.2f}\nAverage: ${average:.2f}...")
    ])

    report_chain = report_prompt | llm
    report_response = await report_chain.ainvoke({
        "product": product_query,
        "median": median_price,
        "average": average_price,
        # ... other parameters
    })

    return {
        **state,
        "final_report": report_response.content,
        "confidence_score": confidence_score
    }

from langgraph.graph import StateGraph

def create_insurance_graph():
    """Create and compile the insurance price analysis graph."""

    workflow = StateGraph(InsuranceState)

    # Add nodes
    workflow.add_node("search_products", search_products)
    workflow.add_node("extract_prices", extract_prices)
    workflow.add_node("generate_report", generate_report)

    # Add edges (linear flow)
    workflow.add_edge("search_products", "extract_prices")
    workflow.add_edge("extract_prices", "generate_report")

    # Set entry and finish points
    workflow.set_entry_point("search_products")
    workflow.set_finish_point("generate_report")

    return workflow.compile()

import streamlit as st
import asyncio

def main():
    st.set_page_config(page_title="🛡️ Product Price Insurance", layout="wide")

    st.markdown("# 🛡️ Product Price Insurance System")
    st.markdown("Find and compare prices across major retailers with AI-powered analysis")

    # Chat interface
    if prompt := st.chat_input("Enter a product name (e.g., 'iPhone 16 256GB')"):
        with st.chat_message("user"):
            st.write(prompt)

        with st.chat_message("assistant"):
            # Progress tracking
            progress_bar = st.progress(0)
            status_text = st.empty()
            results_container = st.container()

            def progress_callback(stage, data, progress_value):
                progress_bar.progress(progress_value)
                if stage == "search_complete":
                    with results_container:
                        display_urls_step(data)
                elif stage == "extract_complete":
                    with results_container:
                        display_prices_step(data)

            # Run analysis with progress updates
            result = asyncio.run(run_analysis_with_progress(prompt, progress_callback))

            # Display final results
            if result.get("final_report"):
                display_final_report(result["final_report"], result.get("confidence_score", 0))

if __name__ == "__main__":
    main()

async def test_search_products():
    """Test the search_products node in isolation."""

    initial_state = {
        "product_query": "iPhone 16 256GB",
        "search_results": {},
        "price_data": [],
        "final_report": "",
        "error": None,
        "confidence_score": None
    }

    result = await search_products(initial_state)

    assert len(result.get('search_results', {})) > 0
    assert result.get('error') is None
    print(f"✅ Found {len(result['search_results'])} URLs")

📱 Product Price Analysis: iPhone 16 256GB

💰 Price Summary:
• Average Price: $899.99
• Median Price: $899.00  
• Price Range: $849.00 - $949.00

🏪 Platform Breakdown:
• Amazon: $899.00 - iPhone 16 256GB Unlocked
• Walmart: $849.00 - Apple iPhone 16 256GB  
• Best Buy: $949.00 - iPhone 16 256GB

🎯 Confidence Score: 8.5/10
✅ 3/3 platforms successfully searched

# ❌ Fragile regex approach
price_match = re.search(r"Price:?\s*(\d+\.?\d*)", content)

# ✅ Reliable structured output
structured_llm = llm.with_structured_output(ExtractedPrice)
result = await structured_llm.ainvoke(content)

pip install langgraph streamlit langchain-google-genai mcp-use python-dotenv

price-shield-agent/
├── state.py          # State definition
├── nodes.py          # Node implementations  
├── graph.py          # Graph builder
├── app.py            # Streamlit interface
└── test_nodes.py     # Testing suite