Chapter 2: First Working Endpoint --- GUVI Compliance

What We Built

This chapter covers the most critical milestone in the project: getting a POST endpoint live that accepts scam messages and returns structured responses in the exact format the GUVI evaluator expects. We will walk through the Pydantic models that define the API contract, the authentication layer, the handler's request-response lifecycle, and the model fallback strategy. We will also discuss the hardest design tension we faced: strict validation vs. lenient parsing.

Why This Approach

The GUVI API Contract

The GUVI evaluator is an automated system. It sends a POST request with a specific JSON structure, expects a response with specific fields, and scores your system based on both the content and the format. Getting the format wrong means scoring zero, regardless of how good your AI is.

Here is the contract, simplified:

sequenceDiagram
    participant E as GUVI Evaluator
    participant H as ScamShield Endpoint
    participant G as Gemini API

    E->>H: POST /guvi_honeypot
    Note over E,H: Headers: x-api-key, Content-Type: application/json
    Note over E,H: Body: { sessionId, message, conversationHistory, metadata }

    H->>H: Validate API key
    H->>H: Parse request into Pydantic model
    H->>G: Classify scam type
    G-->>H: { classification, confidence }
    H->>G: Generate persona response
    G-->>H: response text
    H->>H: Extract evidence (regex)
    H-->>E: 200 OK
    Note over H,E: { status, reply, scamDetected, extractedIntelligence, ... }

The evaluator acts as the scammer. It sends up to 10 messages per session. Each response is scored on reply quality, scam detection, evidence extraction, and intelligence completeness. Crucially, the evaluator also sends a callback URL where we must POST the final intelligence report.

The Request Format

The evaluator sends requests like this:

{
  "sessionId": "eval-session-2025-001",
  "message": {
    "sender": "scammer",
    "text": "Dear customer, your SBI KYC has expired. Update immediately or account will be blocked.",
    "timestamp": 1706000000
  },
  "conversationHistory": [
    {
      "sender": "scammer",
      "text": "Hello, this is SBI customer service.",
      "timestamp": 1705999900
    },
    {
      "sender": "honeypot",
      "text": "Haan ji? Kaun bol raha hai?",
      "timestamp": 1705999950
    }
  ],
  "metadata": {
    "channel": "WhatsApp",
    "language": "English",
    "locale": "IN"
  }
}

And expects a response like this:

{
  "status": "success",
  "reply": "Ek minute ji, chasma lagata hoon. Aap kaun bol rahe ho? Employee ID kya hai?",
  "sessionId": "eval-session-2025-001",
  "scamDetected": true,
  "scamType": "KYC_BANKING",
  "confidenceLevel": 0.92,
  "extractedIntelligence": {
    "bankAccounts": [],
    "upiIds": [],
    "phoneNumbers": [],
    "suspiciousKeywords": ["KYC", "account blocked", "immediately"],
    "phishingLinks": []
  },
  "engagementMetrics": {
    "engagementDurationSeconds": 45.2,
    "totalMessagesExchanged": 3
  },
  "agentNotes": "Type: KYC_BANKING (92%) | Persona: sharma_uncle, Turn: 3"
}

The Code

Pydantic Models: The API Contract in Code

The models in guvi/models.py are the single source of truth for the API contract. Every field name, every type, every default value is defined here and enforced by Pydantic's validation.

class GuviMessage(BaseModel):
    """Single message in the conversation."""

    sender: str  # Accept any sender value (scammer, honeypot, user, bot, etc.)
    text: str
    timestamp: Union[int, str, float]  # Unix timestamp - accept various formats

    @field_validator("timestamp")
    @classmethod
    def validate_timestamp(cls, v: Union[int, str, float]) -> Union[int, float]:
        """Coerce timestamp to numeric and validate reasonable range."""
        try:
            numeric = float(v)
        except (ValueError, TypeError):
            # Try ISO 8601 parsing as fallback
            try:
                dt = datetime.fromisoformat(str(v).replace("Z", "+00:00"))
                numeric = dt.timestamp()
            except (ValueError, TypeError):
                raise ValueError(f"Timestamp must be numeric or ISO 8601, got: {v!r}")
        if numeric < 0:
            raise ValueError(f"Timestamp must be non-negative, got: {numeric}")
        return int(numeric) if numeric == int(numeric) else numeric

The timestamp validator saved us

The GUVI evaluator does not always send timestamps as integers. In different evaluation runs, we observed Unix timestamps as integers, as floating-point numbers, and as ISO 8601 strings. A strict timestamp: int field would have rejected valid requests. The Union[int, str, float] type with a custom validator handles all three formats gracefully.

The request model:

class GuviRequest(BaseModel):
    """GUVI Hackathon incoming request format."""

    sessionId: str = Field(..., description="Unique session identifier from GUVI")
    message: GuviMessage = Field(..., description="The latest scammer message")
    conversationHistory: List[GuviMessage] = Field(
        default_factory=list, description="Previous messages in this session"
    )
    metadata: GuviMetadata = Field(..., description="Channel and language info")
    source: str = Field(default="guvi", description="Request source: 'guvi' or 'testing'")

The conversationHistory field uses default_factory=list rather than a required field. The evaluator sometimes omits it on the first message of a session. If we required it, we would reject valid first-turn requests.

The response model carries the scoring fields:

class GuviResponse(BaseModel):
    """GUVI Hackathon response format."""

    status: Literal["success", "error"] = "success"
    reply: str = Field(..., description="Honeypot AI response to engage scammer")
    sessionId: Optional[str] = Field(default=None)
    scamDetected: Optional[bool] = None
    scamType: Optional[str] = Field(default=None)
    confidenceLevel: Optional[float] = Field(default=None)
    extractedIntelligence: Optional[ExtractedIntelligence] = None
    engagementMetrics: Optional[EngagementMetrics] = None
    totalMessagesExchanged: Optional[int] = Field(default=None)
    engagementDurationSeconds: Optional[float] = Field(default=None)
    agentNotes: Optional[str] = None

Note the totalMessagesExchanged and engagementDurationSeconds fields at the top level. These duplicate data from engagementMetrics. We added them because the GUVI evaluator's scoring script checks for these fields at both the top level and inside engagementMetrics. Including them in both places is redundant but scores higher.

The intelligence model uses camelCase to match the GUVI spec:

class ExtractedIntelligence(BaseModel):
    """Intelligence extracted during the honeypot conversation."""

    bankAccounts: List[str] = Field(default_factory=list)
    upiIds: List[str] = Field(default_factory=list)
    phishingLinks: List[str] = Field(default_factory=list)
    phoneNumbers: List[str] = Field(default_factory=list)
    emailAddresses: List[str] = Field(default_factory=list)
    suspiciousKeywords: List[str] = Field(default_factory=list)
    ifscCodes: List[str] = Field(default_factory=list)
    cryptoWallets: List[str] = Field(default_factory=list)
    aadhaarNumbers: List[str] = Field(default_factory=list)
    panNumbers: List[str] = Field(default_factory=list)
    amounts: List[str] = Field(default_factory=list)
    caseIds: List[str] = Field(default_factory=list)
    policyNumbers: List[str] = Field(default_factory=list)
    orderNumbers: List[str] = Field(default_factory=list)

Every field has default_factory=list. This means a freshly constructed ExtractedIntelligence() with no arguments is valid and serializes to all empty lists. This is critical: we always return intelligence, even when we have not extracted anything yet. The evaluator expects the structure to be present.

API Key Authentication

def validate_api_key(request: https_fn.Request) -> bool:
    """Validate the x-api-key header against our SCAMSHIELD_API_KEY."""
    expected_key = os.environ.get("SCAMSHIELD_API_KEY")
    if not expected_key:
        # In production, deny all if key missing
        if os.environ.get("K_SERVICE"):
            logger.error("SCAMSHIELD_API_KEY not set in production")
            return False
        # In dev, allow all requests
        logger.warning("SCAMSHIELD_API_KEY not set - dev mode")
        return True

    provided_key = request.headers.get("x-api-key", "")
    return provided_key == expected_key

The K_SERVICE environment variable is set automatically by Cloud Run (which underlies Cloud Functions 2nd gen). Its presence tells us we are in production. In dev mode (no K_SERVICE), we skip API key validation so local testing works without configuring secrets.

Security note

The comparison provided_key == expected_key is technically vulnerable to timing attacks. In a production security audit, we upgraded this to use hmac.compare_digest() for constant-time comparison. For a hackathon, the simple equality check was sufficient.

The Handler Lifecycle

The process_honeypot_request function is the core logic, separated from the Firebase decorator for testability:

def process_honeypot_request(request_data: dict) -> dict:
    """Process a honeypot request and return response."""
    try:
        # 1. Parse and validate request
        guvi_request = GuviRequest.model_validate(request_data)

        # 2. Rate limiting
        allowed, reason = check_rate_limit(guvi_request.sessionId)
        if not allowed:
            return GuviResponse(
                status="success",
                reply="Ek minute ruko beta, bahut zyada messages aa rahe hain.",
                sessionId=guvi_request.sessionId,
                scamDetected=False,
                extractedIntelligence=ExtractedIntelligence(),
                engagementMetrics=EngagementMetrics(),
                agentNotes="Rate limited",
            ).model_dump()

        # 3. Get or create session (from Firestore)
        session = get_or_create_session(guvi_request.sessionId, ...)

        # 4. Build conversation context
        conversation = build_conversation_history(
            guvi_request.conversationHistory, guvi_request.message
        )

        # 5. Extract evidence from all scammer messages
        new_evidence_dict = _extract_evidence_from_full_conversation(guvi_request)

        # 6. Cross-session intelligence lookup
        cross_session_match = find_matching_evidence(new_evidence_dict, ...)

        # 7. Orchestrator: classify -> persona -> respond
        orchestrator = Orchestrator(gemini_client=get_gemini_client())
        result = orchestrator.process(session=session, ...)

        # 8-13. Merge evidence, store, compute scores, send callback
        ...

        # 14. Return response
        return GuviResponse(
            status="success",
            reply=result.response,
            sessionId=guvi_request.sessionId,
            scamDetected=scam_detected,
            scamType=result.scam_type,
            confidenceLevel=round(final_confidence, 2),
            extractedIntelligence=merged_evidence,
            engagementMetrics=EngagementMetrics(...),
            agentNotes=agent_notes,
        ).model_dump()

    except Exception as e:
        logger.exception(f"Error processing request: {e}")
        # Always return 200 with stalling reply
        return GuviResponse(
            status="success",
            reply="Ek minute, network slow hai.",
            scamDetected=False,
            extractedIntelligence=ExtractedIntelligence(),
            engagementMetrics=EngagementMetrics(),
            agentNotes=f"Error fallback: {str(e)[:100]}",
        ).model_dump()

The most important detail is the error handler at the bottom. We always return HTTP 200 with a valid response structure, even on internal errors. The GUVI evaluator treats non-200 responses or malformed JSON as a complete failure for that turn. A stalling reply ("Network slow hai, ek minute") that scores low on quality is vastly better than a 500 error that scores zero.

The Model Fallback Strategy

The Gemini client implements a two-tier fallback:

flowchart TD
    A[Incoming request] --> B{Gemini 3 Flash\navailable?}
    B -->|Yes| C[Use Gemini 3 Flash]
    B -->|404 Not Found| D[Switch to Gemini 2.0 Flash]
    D --> E{Gemini 2.0 Flash\navailable?}
    E -->|Yes| F[Use Gemini 2.0 Flash]
    E -->|Error| G{Circuit breaker\nopen?}
    G -->|No| H[Raise exception\n-> error fallback response]
    G -->|Yes| I[Keyword-based\nclassification]

    C --> J[Return response]
    F --> J
    I --> J
    H --> J

Tier 1: Model fallback. Gemini 3 Flash is a preview model. Google occasionally rotates preview models, causing 404 errors. When this happens, the client silently switches to the stable gemini-2.0-flash model. The switch is sticky --- once it falls back, it stays on the fallback model for the lifetime of the function instance.

Tier 2: Circuit breaker. If the Gemini API itself is down (not just a specific model), the circuit breaker opens after 5 consecutive failures. Subsequent requests immediately skip the API call and use keyword-based classification. This returns a reasonable confidence score based on pattern matching (e.g., "KYC" + "account blocked" + "immediately" = KYC_BANKING at 0.7 confidence).

Tier 3: Error fallback. If everything fails, the handler catches the exception and returns a stalling response in character. The scammer (or evaluator) sees a message like "Network slow hai, ek minute" --- plausible from a real person and infinitely better than a stack trace.

Testing with curl

Once deployed, testing the endpoint requires the API key:

curl -X POST \
  https://asia-south1-your-gcp-project-id.cloudfunctions.net/guvi_honeypot \
  -H "Content-Type: application/json" \
  -H "x-api-key: $SCAMSHIELD_API_KEY" \
  -d '{
    "sessionId": "test-001",
    "message": {
      "sender": "scammer",
      "text": "Your SBI KYC has expired. Update now or account blocked.",
      "timestamp": 1706000000
    },
    "conversationHistory": [],
    "metadata": {
      "channel": "WhatsApp",
      "language": "English",
      "locale": "IN"
    }
  }'

The response should include a persona-appropriate reply, scamDetected: true, scamType: "KYC_BANKING", and suspiciousKeywords containing at least "KYC" and "account blocked."

Key Architectural Decision

Strict Pydantic validation vs. lenient parsing.

This was the hardest design tension in the entire project. Pydantic v2 defaults to strict validation: if a field is typed as int, passing a string "123" raises a validation error. This is the right default for internal APIs where you control both sides of the contract.

But we do not control the GUVI evaluator. Across different evaluation runs, we observed:

• Timestamps as integers, floats, and ISO 8601 strings
• conversationHistory present as an empty list, or missing entirely
• sender values as "scammer", "user", "bot", "honeypot", "assistant", and other variations
• Metadata fields with inconsistent casing

We chose lenient parsing with explicit coercion:

• timestamp: Union[int, str, float] with a custom validator that converts any format to numeric
• conversationHistory with default_factory=list so missing fields default gracefully
• sender: str with no enum restriction --- we filter honeypot messages by checking against a set of known labels
• GuviRequest.model_validate(request_data) uses Pydantic's lenient mode by default in v2

The principle: be liberal in what you accept, conservative in what you send. We accept varied input formats without complaint. We send responses in the exact format documented by GUVI, with camelCase field names, consistent types, and all optional fields populated.

The cost of leniency

Lenient parsing means we do not catch malformed requests early. A request with a garbage timestamp that happens to be parseable as a float will be accepted. We mitigated this by logging all parsed requests at DEBUG level so we could inspect what the evaluator actually sent. In a production system, we would add validation warnings (log but don't reject) for unusual input shapes.

What We Learned

Lesson: Always return 200

The GUVI evaluator scores each turn independently. A single 500 error does not just lose points for that turn --- it can cause the evaluator to mark the entire session as failed. Our error handler returns a 200 with a stalling reply ("Network slow hai") for every possible failure mode. This is a general principle for evaluation-facing APIs: graceful degradation beats honest errors.

Lesson: Duplicate fields for compatibility

We spent hours debugging why our evidence extraction score was lower than expected, only to discover the evaluator checked for totalMessagesExchanged at the top level of the response, not inside engagementMetrics. Adding the same data in both places immediately improved our score. When you cannot control the evaluator, err on the side of providing data in every location it might be checked.

Lesson: Separate handler logic from Firebase

The process_honeypot_request function takes a plain dict and returns a plain dict. It knows nothing about Firebase, HTTP requests, or Cloud Functions. This made it testable with simple unit tests --- pass in a dictionary, assert on the output dictionary. The Firebase-specific code (guvi_honeypot) is a thin wrapper that handles CORS, method validation, and serialization. This separation saved us hours of debugging time because we could test the core logic without deploying.

---

Previous: Chapter 1 -- Foundations | Next: Chapter 3 -- Persona Engineering