Middleware

"""Attach documents dynamically based on request context."""
from typing import Any, Dict

async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
    """Attach relevant documents to the agent context."""

    # Get the user's text message
    user_text = ""
    for part in content["parts"]:
        if "text" in part:
            user_text = part["text"]
            break

    # Attach a retrieval document based on topic
    if "pricing" in user_text.lower():
        with open("docs/pricing.pdf", "rb") as f:
            content["parts"].append({
                "data": f.read(),
                "mime_type": "application/pdf"
            })

    elif "technical" in user_text.lower():
        with open("docs/technical-specs.pdf", "rb") as f:
            content["parts"].append({
                "data": f.read(),
                "mime_type": "application/pdf"
            })

    return content

"""Add system context and customer data to requests."""
from typing import Any, Dict
import httpx

async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
    """Prepend system context to the user message."""

    # Read request data from the original connector payload.
    payload = context.get("payload", {})
    customer_id = payload.get("customer_id")

    # Fetch customer data from your API
    customer_info = ""

    if customer_id:
        print(f"Fetching customer data for {customer_id}")
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"https://api.example.com/customers/{customer_id}")
            if resp.status_code == 200:
                data = resp.json()
                customer_info = f"""
Customer: {data['name']}
Plan: {data['plan']}
Account Status: {data['status']}
"""
                print(f"Loaded customer: {data['name']} ({data['plan']})")

    # Prepend context as text part
    content["parts"].insert(0, {
        "text": f"[CUSTOMER CONTEXT]\n{customer_info}\n[END CONTEXT]"
    })

    return content

"""Show the agent only the chat message and keep uploaded files."""
from typing import Any, Dict

async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
    # Webhook body: {"user_id": "u123", "message": "What is in this file?"} plus uploads
    payload = context.get("payload", {})
    context["user_id"] = payload.get("user_id")

    # Replace only the text part; file parts stay in the list
    files = [part for part in content["parts"] if "text" not in part]
    content["parts"] = [{"text": payload.get("message", "")}, *files]

    return content

"""Authentication middleware - stops processing if invalid."""
from typing import Any, Dict
from connic import StopProcessing

async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
    """Validate request before agent runs."""

    # Read the original connector payload. For webhooks, this is the JSON body,
    # form fields, or GET query parameters.
    payload = context.get("payload", {})
    auth_token = payload.get("auth_token")

    if not auth_token:
        raise StopProcessing("Authentication required")

    # Store verified request data in context for prompts and tools.
    context["user_id"] = verify_auth_token(auth_token)

    return content

"""Send analytics to external service after agent completes."""
import httpx
from typing import Any, Dict

async def after(response: str, context: Dict[str, Any]) -> str:
    """Send analytics after agent completes."""

    try:
        async with httpx.AsyncClient() as client:
            await client.post(
                "https://analytics.internal/events",
                json={
                    "event": "agent_run_completed",
                    "run_id": context.get("run_id"),
                    "agent": context.get("agent_name"),
                    "duration_ms": context.get("duration_ms"),
                    "tokens": context.get("token_usage", {}),
                }
            )
    except Exception:
        pass  # Don't fail the request if analytics fails

    return response

"""Send an urgent result through a configured middleware outbound connector."""
from typing import Any, Dict
from connic.tools import send_connector

async def after(response: str, context: Dict[str, Any]) -> str:
    payload = context.get("payload", {})
    if payload.get("priority") == "urgent":
        await send_connector(
            "send_to_ops_slack",
            {"text": response},
        )

    return response

"""Transform response format for specific use cases."""
import json
from typing import Any, Dict

def after(response: str, context: Dict[str, Any]) -> str:
    """Wrap response in a standard API format."""

    return json.dumps({
        "success": True,
        "data": response,
        "metadata": {
            "run_id": context.get("run_id"),
            "processed_at": context.get("timestamp"),
        }
    })