Initial FraudLens AI MCP demo

.env

@@ -0,0 +1,2 @@
+ANTHROPIC_API_KEY=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
+FRAUDLENS_DEBUG=true

README.md

@@ -0,0 +1,22 @@
+# FraudLens AI - Multimodal Fraud Detection Agent
+
+Initial scaffold for the FraudLens AI hackathon project.
+
+## Quick start (dev)
+
+1. Create and activate a virtual environment.
+2. Install dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+3. Create a `.env` file with the required API keys (see `fraudlens_app/config.py`).
+4. Run the main Gradio app:
+   ```bash
+   python -m fraudlens_app.main_app
+   ```
+5. Run the Vision Forensics MCP server (stub):
+   ```bash
+   python -m mcp_servers.vision_forensics.server
+   ```
+
+Further documentation, MCP wiring, and fine-tuning pipeline will be added as we implement features.

app.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import gradio as gr
+
+from fraudlens_app.main_app import build_interface as build_main_interface
+from mcp_servers.vision_forensics.server import (
+    build_vision_forensics_interface,
+)
+from mcp_servers.fraud_intelligence.server import (
+    build_fraud_intel_interface,
+)
+
+
+with gr.Blocks(title="FraudLens AI - MCP Demo") as app:
+    gr.Markdown("# FraudLens AI – Multimodal Fraud Detection Agent")
+    gr.Markdown(
+        "Unified demo Space exposing the main FraudLens agent and two custom "
+        "MCP-style servers (Vision Forensics and Fraud Intelligence)."
+    )
+
+    with gr.Tab("FraudLens Agent"):
+        # Embed the main app layout inside this tab
+        main_iface = build_main_interface()
+
+    with gr.Tab("Vision Forensics MCP"):
+        vision_iface = build_vision_forensics_interface()
+
+    with gr.Tab("Fraud Intelligence MCP"):
+        intel_iface = build_fraud_intel_interface()
+
+
+if __name__ == "__main__":
+    app.launch()

fraudlens_app/config.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Optional
+
+from dotenv import load_dotenv
+
+
+load_dotenv()
+
+
+@dataclass
+class APIConfig:
+    anthropic_api_key: Optional[str] = os.getenv("ANTHROPIC_API_KEY")
+    openai_api_key: Optional[str] = os.getenv("OPENAI_API_KEY")
+    gemini_api_key: Optional[str] = os.getenv("GEMINI_API_KEY")
+    hf_api_token: Optional[str] = os.getenv("HF_API_TOKEN")
+    modal_token_id: Optional[str] = os.getenv("MODAL_TOKEN_ID")
+    modal_token_secret: Optional[str] = os.getenv("MODAL_TOKEN_SECRET")
+    elevenlabs_api_key: Optional[str] = os.getenv("ELEVENLABS_API_KEY")
+
+
+@dataclass
+class AppConfig:
+    debug: bool = os.getenv("FRAUDLENS_DEBUG", "false").lower() == "true"
+    app_title: str = "FraudLens AI - Marketplace Fraud Detection"
+    app_description: str = (
+        "Multimodal fraud detection for second-hand marketplace listings "
+        "using vision-language models and Chain-of-Thought reasoning."
+    )
+
+
+api_config = APIConfig()
+app_config = AppConfig()

fraudlens_app/main_app.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+import gradio as gr
+
+from .config import app_config, api_config
+from .reasoning import ReasoningEngine
+from mcp_servers.vision_forensics.server import (
+    detect_image_manipulation,
+    extract_image_metadata,
+    reverse_image_search,
+    compare_image_quality,
+    detect_ai_generated_image,
+)
+
+
+def build_interface() -> gr.Blocks:
+    with gr.Blocks(title=app_config.app_title) as demo:
+        gr.Markdown(f"# {app_config.app_title}")
+        gr.Markdown(app_config.app_description)
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                image_input = gr.Image(label="Listing Images", type="filepath")
+                description_input = gr.Textbox(
+                    label="Listing Description",
+                    placeholder="Paste the marketplace ad description...",
+                    lines=5,
+                )
+                price_input = gr.Textbox(label="Price", placeholder="e.g. 120.00")
+                category_input = gr.Textbox(label="Category", placeholder="e.g. Smartphone")
+                seller_info_input = gr.Textbox(
+                    label="Seller Info",
+                    placeholder="Seller ID, rating, history, etc.",
+                    lines=3,
+                )
+                analyze_btn = gr.Button("Analyze Fraud Risk", variant="primary")
+
+            with gr.Column(scale=1):
+                risk_score_output = gr.Number(label="Risk Score (0-100)")
+                reasoning_output = gr.JSON(label="Reasoning Chain")
+
+        status_output = gr.Markdown("Ready.")
+
+        def _analyze(image_path: str, desc: str, price: str, category: str, seller: str) -> Dict[str, Any]:
+            if not api_config.anthropic_api_key:
+                return 0, {"error": "ANTHROPIC_API_KEY not configured"}, "Missing Anthropic API key."
+
+            try:
+                vision_signals: Dict[str, Any] = {}
+                if image_path:
+                    # Call local Vision Forensics tools; in a full MCP setup these
+                    # would be remote tool calls instead of direct imports.
+                    vision_signals = {
+                        "manipulation": detect_image_manipulation(image_path),
+                        "metadata": extract_image_metadata(image_path),
+                        "reverse_search": reverse_image_search(image_path),
+                        "ai_generated": detect_ai_generated_image(image_path),
+                        # category-aware quality placeholder
+                        "quality": compare_image_quality(image_path, category or ""),
+                    }
+
+                payload: Dict[str, Any] = {
+                    "image_path": image_path,
+                    "description": desc,
+                    "price": price,
+                    "category": category,
+                    "seller_info": seller,
+                    "vision_forensics": vision_signals,
+                }
+
+                engine = ReasoningEngine()
+                result = engine.analyze_listing(payload)
+                # Pass through structured reasoning (steps, confidence, flags, etc.)
+                structured = {
+                    "steps": result.get("steps", {}),
+                    "risk_score": result.get("risk_score", 0),
+                    "confidence": result.get("confidence", 0.0),
+                    "flags": result.get("flags", []),
+                }
+                return (
+                    structured["risk_score"],
+                    structured,
+                    "Analysis complete.",
+                )
+            except Exception as exc:  # pylint: disable=broad-except
+                return 0, {"error": str(exc)}, "Error during analysis. Check logs."
+
+        analyze_btn.click(
+            _analyze,
+            inputs=[image_input, description_input, price_input, category_input, seller_info_input],
+            outputs=[risk_score_output, reasoning_output, status_output],
+        )
+
+    return demo
+
+
+if __name__ == "__main__":
+    iface = build_interface()
+    iface.launch()

fraudlens_app/reasoning.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+import json
+
+import anthropic
+
+from .config import api_config
+
+
+class ReasoningEngine:
+    """Wrapper around Anthropic Claude for structured Chain-of-Thought reasoning.
+
+    Enforces a 4-step JSON schema:
+
+    {
+      "steps": {
+        "visual_analysis": {...},
+        "textual_analysis": {...},
+        "cross_validation": {...},
+        "risk_scoring": {...}
+      },
+      "risk_score": int,           # 0-100
+      "confidence": float,         # 0-1
+      "flags": [ {"type": str, "severity": str, "message": str} ]
+    }
+    """
+
+    def __init__(self) -> None:
+        if not api_config.anthropic_api_key:
+            raise RuntimeError("ANTHROPIC_API_KEY is not set")
+
+        # Use official Anthropic client for correct base URL, headers, and
+        # versioning.
+        self._client = anthropic.Anthropic(api_key=api_config.anthropic_api_key)
+
+    def _default_response(self, raw_text: str | None, raw: Dict[str, Any]) -> Dict[str, Any]:
+        """Fallback response if Claude does not return valid JSON."""
+
+        reasoning = raw_text or "Model response missing or unparsable; using fallback schema."
+        return {
+            "steps": {
+                "visual_analysis": {
+                    "summary": "Not available - JSON parse failed.",
+                    "signals": [],
+                },
+                "textual_analysis": {
+                    "summary": "Not available - JSON parse failed.",
+                    "signals": [],
+                },
+                "cross_validation": {
+                    "summary": "Not available - JSON parse failed.",
+                    "inconsistencies": [],
+                },
+                "risk_scoring": {
+                    "summary": reasoning,
+                    "score": 50,
+                    "confidence": 0.3,
+                    "explanation": reasoning,
+                },
+            },
+            "risk_score": 50,
+            "confidence": 0.3,
+            "flags": [
+                {
+                    "type": "model_output_error",
+                    "severity": "medium",
+                    "message": "Claude did not return valid JSON in the expected schema.",
+                }
+            ],
+            "raw_response": raw,
+        }
+
+    def analyze_listing(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """Call Claude with a strict JSON schema for listing analysis."""
+
+        schema_instructions = """
+You MUST respond with a single JSON object only, no prose, following this schema:
+{
+  "steps": {
+    "visual_analysis": {
+      "summary": "string - short explanation of visual findings",
+      "signals": [
+        {"name": "string", "weight": "low|medium|high", "evidence": "string"}
+      ]
+    },
+    "textual_analysis": {
+      "summary": "string - short explanation of textual findings",
+      "signals": [
+        {"name": "string", "weight": "low|medium|high", "evidence": "string"}
+      ]
+    },
+    "cross_validation": {
+      "summary": "string - how visuals, text, price, and seller info align or conflict",
+      "inconsistencies": [
+        {"description": "string", "impact": "low|medium|high"}
+      ]
+    },
+    "risk_scoring": {
+      "summary": "string - overall rationale",
+      "score": 0-100,
+      "confidence": 0.0-1.0,
+      "explanation": "string - concise reasoning chain"
+    }
+  },
+  "risk_score": 0-100,
+  "confidence": 0.0-1.0,
+  "flags": [
+    {"type": "string", "severity": "low|medium|high|critical", "message": "string"}
+  ]
+}
+"""
+
+        system_prompt = (
+            "You are FraudLens AI, an expert fraud detection agent for "
+            "second-hand marketplaces. Perform a 4-step analysis: "
+            "(1) visual analysis, (2) textual analysis, (3) cross-validation "
+            "between visuals/text/price/seller, and (4) risk scoring. "
+            "Return ONLY valid JSON, no markdown, following the provided schema."
+        )
+
+        user_content = (
+            "Analyze the following marketplace listing and respond strictly "
+            "with JSON in the required schema. Listing payload (JSON):\n" f"{payload}\n\n" + schema_instructions
+        )
+
+        try:
+            msg = self._client.messages.create(
+                # Use a stable, current Claude Sonnet alias
+                model="claude-3-7-sonnet-latest",
+                max_tokens=900,
+                temperature=0.2,
+                system=system_prompt,
+                messages=[
+                    {"role": "user", "content": user_content},
+                ],
+            )
+        except Exception as exc:  # pragma: no cover - network / API errors
+            # Surface the error in a structured way to the UI
+            return {
+                "steps": {},
+                "risk_score": 0,
+                "confidence": 0.0,
+                "flags": [
+                    {
+                        "type": "anthropic_error",
+                        "severity": "high",
+                        "message": str(exc),
+                    }
+                ],
+                "raw_response": {},
+            }
+
+        # Anthropic SDK returns an object with .content as a list of blocks
+        data: Dict[str, Any] = msg.model_dump()  # for logging / debugging
+
+        raw_text: str | None = None
+        if msg.content and isinstance(msg.content, list):
+            first_block = msg.content[0]
+            # For TextBlock, attribute is .text
+            raw_text = getattr(first_block, "text", None)
+
+        if not raw_text:
+            return self._default_response(None, data)
+
+        try:
+            parsed = json.loads(raw_text)
+        except json.JSONDecodeError:
+            return self._default_response(raw_text, data)
+
+        # Ensure top-level keys exist; fill safe defaults if missing
+        parsed.setdefault("steps", {})
+        steps = parsed["steps"]
+        steps.setdefault("visual_analysis", {"summary": "", "signals": []})
+        steps.setdefault("textual_analysis", {"summary": "", "signals": []})
+        steps.setdefault("cross_validation", {"summary": "", "inconsistencies": []})
+        steps.setdefault(
+            "risk_scoring",
+            {"summary": "", "score": parsed.get("risk_score", 50), "confidence": parsed.get("confidence", 0.5), "explanation": ""},
+        )
+
+        parsed.setdefault("risk_score", steps["risk_scoring"].get("score", 50))
+        parsed.setdefault("confidence", steps["risk_scoring"].get("confidence", 0.5))
+        parsed.setdefault("flags", [])
+
+        parsed["raw_response"] = data
+        return parsed

mcp_servers/fraud_intelligence/server.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+import gradio as gr
+
+
+def check_price_anomaly(price: float, category: str, condition: str) -> Dict[str, Any]:
+    # TODO: replace with real price benchmark logic
+    return {
+        "is_anomalous": False,
+        "z_score": 0.0,
+        "benchmark_range": {
+            "min": None,
+            "max": None,
+        },
+        "details": "Stub price anomaly check.",
+    }
+
+
+def analyze_seller_history(seller_id: str) -> Dict[str, Any]:
+    # TODO: integrate with real seller history / reputation data
+    return {
+        "seller_id": seller_id,
+        "risk_level": "unknown",
+        "metrics": {
+            "total_listings": 0,
+            "fraud_reports": 0,
+            "account_age_days": None,
+        },
+        "details": "Stub seller history analysis.",
+    }
+
+
+def detect_listing_patterns(seller_id: str, timeframe_days: int = 30) -> Dict[str, Any]:
+    # TODO: detect velocity, category concentration, repeated templates, etc.
+    return {
+        "seller_id": seller_id,
+        "timeframe_days": timeframe_days,
+        "patterns": [],
+        "details": "Stub listing pattern detection.",
+    }
+
+
+def cross_reference_listings(ad_content: str) -> Dict[str, Any]:
+    # TODO: cross-reference against known fraud corpus / duplicate listings
+    return {
+        "matches": [],
+        "details": "Stub cross-reference - no external corpus yet.",
+    }
+
+
+def get_fraud_risk_score(ad_data: Dict[str, Any]) -> Dict[str, Any]:
+    # TODO: combine multiple intelligence signals into a composite score
+    return {
+        "risk_score": 0,
+        "confidence": 0.0,
+        "factors": [],
+        "details": "Stub fraud risk scorer.",
+    }
+
+
+def build_fraud_intel_interface() -> gr.Blocks:
+    with gr.Blocks(title="Fraud Intelligence MCP Server") as demo:
+        gr.Markdown("# Fraud Intelligence MCP Server (Stub)")
+        gr.Markdown(
+            "Tools for price anomaly detection, seller reputation, and pattern "
+            "analysis. This Gradio app will later be wrapped as an MCP server."
+        )
+
+        with gr.Tab("Price Anomaly"):
+            price_in = gr.Number(label="Price")
+            cat_in = gr.Textbox(label="Category")
+            cond_in = gr.Textbox(label="Condition")
+            price_btn = gr.Button("Check Price")
+            price_out = gr.JSON(label="Price Anomaly Result")
+            price_btn.click(
+                check_price_anomaly,
+                inputs=[price_in, cat_in, cond_in],
+                outputs=[price_out],
+            )
+
+        with gr.Tab("Seller History"):
+            seller_id_in = gr.Textbox(label="Seller ID")
+            seller_btn = gr.Button("Analyze Seller")
+            seller_out = gr.JSON(label="Seller History")
+            seller_btn.click(analyze_seller_history, inputs=[seller_id_in], outputs=[seller_out])
+
+        with gr.Tab("Listing Patterns"):
+            seller_id_pat = gr.Textbox(label="Seller ID")
+            timeframe_in = gr.Number(label="Timeframe (days)", value=30)
+            pattern_btn = gr.Button("Detect Patterns")
+            pattern_out = gr.JSON(label="Patterns")
+            pattern_btn.click(
+                detect_listing_patterns,
+                inputs=[seller_id_pat, timeframe_in],
+                outputs=[pattern_out],
+            )
+
+        with gr.Tab("Cross Reference"):
+            ad_content_in = gr.Textbox(label="Ad Content", lines=4)
+            cross_btn = gr.Button("Cross Reference Listings")
+            cross_out = gr.JSON(label="Cross Reference Result")
+            cross_btn.click(cross_reference_listings, inputs=[ad_content_in], outputs=[cross_out])
+
+    return demo
+
+
+if __name__ == "__main__":
+    app = build_fraud_intel_interface()
+    app.launch()

mcp_servers/vision_forensics/server.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+import gradio as gr
+
+
+def detect_image_manipulation(image_path: str) -> Dict[str, Any]:
+    # TODO: integrate OpenCV / ML-based manipulation detection
+    return {
+        "manipulated": False,
+        "confidence": 0.0,
+        "details": "Stub implementation - analysis not yet implemented.",
+    }
+
+
+def extract_image_metadata(image_path: str) -> Dict[str, Any]:
+    # TODO: add EXIF parsing and metadata checks
+    return {"metadata": {}, "details": "Stub metadata extractor."}
+
+
+def reverse_image_search(image_path: str) -> Dict[str, Any]:
+    # TODO: integrate with a reverse image search API/service
+    return {"matches": [], "details": "Stub reverse image search."}
+
+
+def compare_image_quality(image_path: str, category: str) -> Dict[str, Any]:
+    # TODO: implement quality heuristics per category
+    return {"quality_score": 0.5, "details": "Stub quality comparison."}
+
+
+def detect_ai_generated_image(image_path: str) -> Dict[str, Any]:
+    # TODO: integrate with AI-generated image detection model
+    return {"ai_generated": False, "confidence": 0.0, "details": "Stub AI detection."}
+
+
+def build_vision_forensics_interface() -> gr.Blocks:
+    with gr.Blocks(title="Vision Forensics MCP Server") as demo:
+        gr.Markdown("# Vision Forensics MCP Server (Stub)")
+        gr.Markdown(
+            "Exposes image forensics tools for FraudLens AI. "
+            "This Gradio app will later be wrapped as an MCP server."
+        )
+
+        with gr.Tab("Manipulation Detection"):
+            img_in = gr.Image(type="filepath")
+            btn = gr.Button("Detect Manipulation")
+            out = gr.JSON(label="Result")
+            btn.click(detect_image_manipulation, inputs=[img_in], outputs=[out])
+
+        with gr.Tab("Metadata"):
+            img_meta_in = gr.Image(type="filepath")
+            meta_btn = gr.Button("Extract Metadata")
+            meta_out = gr.JSON(label="Metadata")
+            meta_btn.click(extract_image_metadata, inputs=[img_meta_in], outputs=[meta_out])
+
+        with gr.Tab("Reverse Search"):
+            img_rev_in = gr.Image(type="filepath")
+            rev_btn = gr.Button("Reverse Search")
+            rev_out = gr.JSON(label="Matches")
+            rev_btn.click(reverse_image_search, inputs=[img_rev_in], outputs=[rev_out])
+
+        with gr.Tab("AI Detection"):
+            img_ai_in = gr.Image(type="filepath")
+            ai_btn = gr.Button("Detect AI")
+            ai_out = gr.JSON(label="AI Detection Result")
+            ai_btn.click(detect_ai_generated_image, inputs=[img_ai_in], outputs=[ai_out])
+
+    return demo
+
+
+if __name__ == "__main__":
+    app = build_vision_forensics_interface()
+    app.launch()

requirements.txt

@@ -0,0 +1,11 @@
+gradio>=4.0.0
+anthropic>=0.34.0
+openai>=1.0.0
+google-generativeai>=0.8.0
+httpx>=0.27.0
+pydantic>=2.7.0
+python-dotenv>=1.0.1
+opencv-python>=4.10.0.84
+Pillow>=10.0.0
+modal>=0.64.0
+huggingface_hub>=0.24.0