Production-Ready Agents: Observability, Security & Deployment – Part 8

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.aiohttp_client import AioHttpClientInstrumentor
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def configure_telemetry(service_name: str = "agent-service"):
    """Configure OpenTelemetry for the agent service."""
    
    # Create resource with service metadata
    resource = Resource.create({
        "service.name": service_name,
        "service.version": "1.0.0",
        "deployment.environment": os.getenv("ENVIRONMENT", "development")
    })
    
    # Create tracer provider
    provider = TracerProvider(resource=resource)
    
    # Configure OTLP exporter (sends to collector)
    otlp_exporter = OTLPSpanExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
        insecure=True
    )
    
    # Add batch processor for efficient export
    processor = BatchSpanProcessor(otlp_exporter)
    provider.add_span_processor(processor)
    
    # Set as global tracer provider
    trace.set_tracer_provider(provider)
    
    # Instrument HTTP client
    AioHttpClientInstrumentor().instrument()
    
    logger.info(f"Telemetry configured for {service_name}")
    return trace.get_tracer(service_name)

# Custom span decorator for agent operations
def traced(operation_name: str):
    """Decorator to trace agent operations."""
    def decorator(func):
        async def wrapper(*args, **kwargs):
            tracer = trace.get_tracer(__name__)
            with tracer.start_as_current_span(operation_name) as span:
                span.set_attribute("agent.operation", operation_name)
                try:
                    result = await func(*args, **kwargs)
                    span.set_attribute("agent.success", True)
                    return result
                except Exception as e:
                    span.set_attribute("agent.success", False)
                    span.set_attribute("agent.error", str(e))
                    span.record_exception(e)
                    raise
        return wrapper
    return decorator

# Usage example
tracer = configure_telemetry("my-agent-service")

@traced("process_user_request")
async def handle_request(message: str, thread):
    result = await agent.run(message, thread)
    return result

from azure.identity import (
    DefaultAzureCredential,
    ManagedIdentityCredential,
    ChainedTokenCredential,
    AzureCliCredential
)
import os
import logging

logger = logging.getLogger(__name__)

def get_azure_credential():
    """
    Get appropriate Azure credential based on environment.
    
    Priority:
    1. Managed Identity (when running in Azure)
    2. Azure CLI (for local development)
    3. Default credential chain (fallback)
    """
    
    # Check if running in Azure
    azure_client_id = os.getenv("AZURE_CLIENT_ID")
    
    if azure_client_id:
        # Production: Use Managed Identity
        logger.info("Using Managed Identity authentication")
        return ManagedIdentityCredential(client_id=azure_client_id)
    
    # Development: Use Azure CLI first, then fallback
    logger.info("Using development credential chain")
    return ChainedTokenCredential(
        AzureCliCredential(),
        DefaultAzureCredential()
    )

# Secure agent initialization
def create_secure_agent():
    """Create agent with secure credential management."""
    from agent_framework.azure import AzureOpenAIResponsesClient
    
    credential = get_azure_credential()
    
    # Get endpoint from environment (never hardcode)
    endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
    if not endpoint:
        raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
    
    client = AzureOpenAIResponsesClient(
        credential=credential,
        endpoint=endpoint
    )
    
    return client.create_agent(
        name="SecureAgent",
        instructions="You are a helpful assistant."
    )

# Key Vault integration for secrets
from azure.keyvault.secrets import SecretClient

class SecureConfigProvider:
    """Fetch configuration from Azure Key Vault."""
    
    def __init__(self, vault_url: str):
        credential = get_azure_credential()
        self.client = SecretClient(vault_url=vault_url, credential=credential)
    
    def get_secret(self, name: str) -> str:
        """Retrieve a secret from Key Vault."""
        secret = self.client.get_secret(name)
        return secret.value
    
    def get_api_key(self, service: str) -> str:
        """Get API key for external service."""
        return self.get_secret(f"{service}-api-key")

# Usage
config = SecureConfigProvider(os.getenv("AZURE_KEY_VAULT_URL"))
crm_api_key = config.get_api_key("crm")

from dataclasses import dataclass
from typing import List, Optional, Set
from enum import Enum
import re
import logging

logger = logging.getLogger(__name__)

class SafetyCategory(Enum):
    HARMFUL = "harmful"
    PII = "pii"
    JAILBREAK = "jailbreak"
    PROFANITY = "profanity"
    BLOCKED_TERM = "blocked_term"

@dataclass
class SafetyResult:
    is_safe: bool
    violations: List[SafetyCategory]
    details: str

class ContentSafetyFilter:
    """
    Enterprise-grade content safety filter for agent inputs/outputs.
    """
    
    def __init__(
        self,
        block_harmful: bool = True,
        block_pii: bool = True,
        block_jailbreaks: bool = True,
        custom_blocklist: Optional[List[str]] = None,
        max_input_length: int = 4000
    ):
        self.block_harmful = block_harmful
        self.block_pii = block_pii
        self.block_jailbreaks = block_jailbreaks
        self.blocklist: Set[str] = set(custom_blocklist or [])
        self.max_input_length = max_input_length
        
        # PII patterns
        self.pii_patterns = [
            (r'\d{3}-\d{2}-\d{4}', 'SSN'),  # Social Security Number
            (r'\d{16}', 'Credit Card'),      # Credit card number
            (r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}', 'Email'),
            (r'\d{3}[-.]?\d{3}[-.]?\d{4}', 'Phone'),
        ]
        
        # Jailbreak patterns
        self.jailbreak_patterns = [
            r'ignore (previous|all|your) instructions',
            r'pretend (you are|to be)',
            r'act as (if you are|a)',
            r'disregard (safety|guidelines)',
            r'bypass (filters|safety)',
        ]
    
    def check_input(self, text: str) -> SafetyResult:
        """Check user input for safety violations."""
        violations = []
        details = []
        
        # Length check
        if len(text) > self.max_input_length:
            violations.append(SafetyCategory.HARMFUL)
            details.append(f"Input exceeds max length ({len(text)} > {self.max_input_length})")
        
        # PII check
        if self.block_pii:
            for pattern, pii_type in self.pii_patterns:
                if re.search(pattern, text, re.IGNORECASE):
                    violations.append(SafetyCategory.PII)
                    details.append(f"Potential {pii_type} detected")
        
        # Jailbreak check
        if self.block_jailbreaks:
            for pattern in self.jailbreak_patterns:
                if re.search(pattern, text, re.IGNORECASE):
                    violations.append(SafetyCategory.JAILBREAK)
                    details.append("Potential jailbreak attempt detected")
                    break
        
        # Blocklist check
        text_lower = text.lower()
        for term in self.blocklist:
            if term.lower() in text_lower:
                violations.append(SafetyCategory.BLOCKED_TERM)
                details.append(f"Blocked term detected")
        
        is_safe = len(violations) == 0
        
        if not is_safe:
            logger.warning(f"Content safety violation: {details}")
        
        return SafetyResult(
            is_safe=is_safe,
            violations=list(set(violations)),
            details="; ".join(details) if details else "No issues detected"
        )
    
    def sanitize_output(self, text: str) -> str:
        """Sanitize agent output by redacting PII."""
        result = text
        
        for pattern, pii_type in self.pii_patterns:
            result = re.sub(pattern, f"[{pii_type} REDACTED]", result)
        
        return result

# Usage with agent
class SafeAgent:
    def __init__(self, agent, filter_config: Optional[dict] = None):
        self.agent = agent
        self.filter = ContentSafetyFilter(**(filter_config or {}))
    
    async def run(self, message: str, thread=None):
        # Check input safety
        safety_check = self.filter.check_input(message)
        
        if not safety_check.is_safe:
            return f"I cannot process this request: {safety_check.details}"
        
        # Run agent
        result = await self.agent.run(message, thread)
        
        # Sanitize output
        sanitized = self.filter.sanitize_output(result.text)
        
        return sanitized

import asyncio
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Callable
import logging

logger = logging.getLogger(__name__)

@dataclass
class CircuitBreakerState:
    failures: int = 0
    last_failure: Optional[datetime] = None
    is_open: bool = False
    
class ResilientAgent:
    """
    Production-ready agent wrapper with:
    - Automatic retry with exponential backoff
    - Circuit breaker for failure protection
    - Timeout handling
    - Fallback responses
    """
    
    def __init__(
        self,
        agent,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 30.0,
        timeout: float = 60.0,
        circuit_threshold: int = 5,
        circuit_reset_time: int = 60,
        fallback_response: Optional[str] = None
    ):
        self.agent = agent
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.timeout = timeout
        self.circuit_threshold = circuit_threshold
        self.circuit_reset_time = circuit_reset_time
        self.fallback_response = fallback_response or "I'm experiencing difficulties. Please try again later."
        
        self.circuit = CircuitBreakerState()
    
    def _check_circuit(self) -> bool:
        """Check if circuit breaker allows requests."""
        if not self.circuit.is_open:
            return True
        
        # Check if reset time has passed
        if self.circuit.last_failure:
            elapsed = datetime.now() - self.circuit.last_failure
            if elapsed > timedelta(seconds=self.circuit_reset_time):
                logger.info("Circuit breaker reset - allowing requests")
                self.circuit.is_open = False
                self.circuit.failures = 0
                return True
        
        logger.warning("Circuit breaker is OPEN - rejecting request")
        return False
    
    def _record_failure(self):
        """Record a failure and potentially open the circuit."""
        self.circuit.failures += 1
        self.circuit.last_failure = datetime.now()
        
        if self.circuit.failures >= self.circuit_threshold:
            self.circuit.is_open = True
            logger.error(f"Circuit breaker OPENED after {self.circuit.failures} failures")
    
    def _record_success(self):
        """Record a success and reset failure count."""
        self.circuit.failures = 0
    
    async def run(
        self,
        message: str,
        thread=None,
        on_retry: Optional[Callable] = None
    ) -> str:
        """
        Run agent with resilience patterns.
        """
        # Check circuit breaker
        if not self._check_circuit():
            return self.fallback_response
        
        last_error = None
        
        for attempt in range(self.max_retries + 1):
            try:
                # Apply timeout
                result = await asyncio.wait_for(
                    self.agent.run(message, thread),
                    timeout=self.timeout
                )
                
                self._record_success()
                return result.text
                
            except asyncio.TimeoutError:
                last_error = "Request timed out"
                logger.warning(f"Attempt {attempt + 1}: Timeout after {self.timeout}s")
                
            except Exception as e:
                last_error = str(e)
                logger.warning(f"Attempt {attempt + 1}: {last_error}")
            
            # Record failure
            self._record_failure()
            
            # If not last attempt, wait with exponential backoff
            if attempt < self.max_retries:
                delay = min(self.base_delay * (2 ** attempt), self.max_delay)
                logger.info(f"Retrying in {delay:.1f} seconds...")
                
                if on_retry:
                    on_retry(attempt + 1, delay)
                
                await asyncio.sleep(delay)
        
        # All retries exhausted
        logger.error(f"All retries exhausted. Last error: {last_error}")
        return self.fallback_response

# Usage
resilient_agent = ResilientAgent(
    agent=base_agent,
    max_retries=3,
    timeout=45.0,
    circuit_threshold=5,
    fallback_response="Our AI assistant is temporarily unavailable. Please try again in a few minutes."
)

result = await resilient_agent.run("Help me with my order")

import os
from azure.identity import DefaultAzureCredential
from agent_framework.azure import AzureAIAgentClient

async def deploy_to_azure_ai_foundry():
    """
    Deploy agent to Azure AI Foundry Agent Service.
    
    Benefits:
    - Fully managed infrastructure
    - Automatic scaling
    - Built-in monitoring and logging
    - Enterprise security controls
    - SLA-backed reliability
    """
    
    # Get project connection string from Azure AI Foundry
    project_connection = os.getenv("AZURE_AI_PROJECT_CONNECTION")
    
    if not project_connection:
        raise ValueError("AZURE_AI_PROJECT_CONNECTION required for Azure deployment")
    
    # Create Azure AI Agent client
    client = AzureAIAgentClient(
        project_connection_string=project_connection,
        credential=DefaultAzureCredential()
    )
    
    # Define agent configuration
    agent_config = {
        "name": "ProductionSupportAgent",
        "instructions": """
            You are an enterprise customer support agent.
            - Be professional and helpful
            - Use tools to look up information
            - Escalate complex issues appropriately
            - Always protect customer privacy
        """,
        "model": "gpt-4o",
        "tools": [
            {"type": "code_interpreter"},  # Built-in code execution
            {"type": "file_search"}         # RAG over documents
        ]
    }
    
    # Create agent in Azure AI Foundry
    agent = await client.create_agent(**agent_config)
    
    print(f"Agent deployed to Azure AI Foundry")
    print(f"  Agent ID: {agent.id}")
    print(f"  Endpoint: {agent.endpoint}")
    
    # Agent is now managed by Azure:
    # - Auto-scaling based on demand
    # - Automatic failover
    # - Integrated with Azure Monitor
    # - RBAC security model
    
    return agent

# Using the deployed agent
async def use_foundry_agent():
    agent = await deploy_to_azure_ai_foundry()
    
    # Create thread for conversation
    thread = await agent.create_thread()
    
    # Run conversation (fully managed by Azure)
    result = await agent.run(
        thread_id=thread.id,
        message="I need help with my subscription",
        stream=True  # Stream responses
    )
    
    async for chunk in result:
        print(chunk.text, end="", flush=True)
    
    print(f"\n\nConversation ID: {thread.id}")

# Run deployment
asyncio.run(use_foundry_agent())

# Dockerfile for production agent service
FROM python:3.11-slim as base

# Security: Run as non-root user
RUN useradd -m -u 1000 agent
WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY --chown=agent:agent . .

# Switch to non-root user
USER agent

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

# Run with uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

---
# docker-compose.yml for local development
version: '3.8'

services:
  agent-service:
    build: .
    ports:
      - "8000:8000"
    environment:
      - AZURE_OPENAI_ENDPOINT=${AZURE_OPENAI_ENDPOINT}
      - AZURE_CLIENT_ID=${AZURE_CLIENT_ID}
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
      - otel-collector
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data

  otel-collector:
    image: otel/opentelemetry-collector-contrib:latest
    command: ["--config=/etc/otel-collector-config.yaml"]
    volumes:
      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
      - "8889:8889"   # Prometheus metrics

  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"  # Jaeger UI
      - "14250:14250"  # gRPC

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana-data:/var/lib/grafana

volumes:
  redis-data:
  grafana-data: