"""
Pattern 122: Client-Server Pattern

Purpose: Request-response architecture with centralized service coordination
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


class RequestType(Enum):
    """Types of client requests."""
    QUERY = "query"
    COMMAND = "command"
    STREAM = "stream"


@dataclass
class ClientRequest:
    """Client request to server."""
    client_id: str
    request_type: RequestType
    payload: Any
    request_id: str = field(default_factory=lambda: f"req_{datetime.now().timestamp()}")
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ServerResponse:
    """Server response to client."""
    request_id: str
    status: str  # success, error, pending
    data: Any
    server_id: str
    processing_time: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)


class Client:
    """Client that sends requests to server."""
    
    def __init__(self, client_id: str, model_name: str = "gpt-4"):
        self.client_id = client_id
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.requests_sent: List[ClientRequest] = []
        self.responses_received: List[ServerResponse] = []
    
    def send_request(self, request_type: RequestType, payload: Any) -> ClientRequest:
        """Send request to server."""
        request = ClientRequest(
            client_id=self.client_id,
            request_type=request_type,
            payload=payload
        )
        self.requests_sent.append(request)
        return request
    
    def receive_response(self, response: ServerResponse):
        """Receive response from server."""
        self.responses_received.append(response)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get client statistics."""
        success_count = sum(1 for r in self.responses_received if r.status == "success")
        return {
            "client_id": self.client_id,
            "requests_sent": len(self.requests_sent),
            "responses_received": len(self.responses_received),
            "success_rate": success_count / len(self.responses_received) if self.responses_received else 0.0
        }


class Server:
    """Central server that processes client requests."""
    
    def __init__(self, server_id: str, model_name: str = "gpt-4"):
        self.server_id = server_id
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.requests_processed: List[ClientRequest] = []
        self.active_connections: Dict[str, Client] = {}
    
    def register_client(self, client: Client):
        """Register a client connection."""
        self.active_connections[client.client_id] = client
    
    def process_request(self, request: ClientRequest) -> ServerResponse:
        """Process client request."""
        start_time = datetime.now()
        
        # Use LLM to process request based on type
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a server processing {request.request_type.value} requests."),
            ("human", "Client {client_id} requests: {payload}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "client_id": request.client_id,
            "payload": str(request.payload)
        })
        
        end_time = datetime.now()
        processing_time = (end_time - start_time).total_seconds()
        
        self.requests_processed.append(request)
        
        return ServerResponse(
            request_id=request.request_id,
            status="success",
            data=response.content,
            server_id=self.server_id,
            processing_time=processing_time
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get server statistics."""
        avg_processing_time = (
            sum(r.processing_time for r in self.requests_processed) / len(self.requests_processed)
            if self.requests_processed else 0.0
        )
        return {
            "server_id": self.server_id,
            "active_clients": len(self.active_connections),
            "requests_processed": len(self.requests_processed),
            "avg_processing_time": avg_processing_time
        }


class ClientServerPattern:
    """
    Client-Server Pattern - Centralized request-response architecture
    
    Implements client-server communication with request routing and response management.
    """
    
    def __init__(self):
        self.servers: Dict[str, Server] = {}
        self.clients: Dict[str, Client] = {}
        self.message_log: List[Dict[str, Any]] = []
    
    def register_server(self, server: Server):
        """Register a server."""
        self.servers[server.server_id] = server
    
    def register_client(self, client: Client):
        """Register a client."""
        self.clients[client.client_id] = client
    
    def route_request(self, client_id: str, server_id: str, 
                     request_type: RequestType, payload: Any) -> ServerResponse:
        """Route client request to server."""
        client = self.clients.get(client_id)
        server = self.servers.get(server_id)
        
        if not client or not server:
            raise ValueError("Invalid client or server ID")
        
        # Client sends request
        request = client.send_request(request_type, payload)
        
        # Server processes request
        response = server.process_request(request)
        
        # Client receives response
        client.receive_response(response)
        
        # Log the transaction
        self.message_log.append({
            "client_id": client_id,
            "server_id": server_id,
            "request_id": request.request_id,
            "request_type": request_type.value,
            "status": response.status,
            "timestamp": datetime.now()
        })
        
        return response
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_servers": len(self.servers),
            "total_clients": len(self.clients),
            "total_requests": len(self.message_log),
            "servers": {sid: server.get_statistics() for sid, server in self.servers.items()},
            "clients": {cid: client.get_statistics() for cid, client in self.clients.items()}
        }


# Example usage
if __name__ == "__main__":
    print("=== Client-Server Pattern Demo ===\n")
    
    # Create pattern
    pattern = ClientServerPattern()
    
    # Create server
    server = Server("MainServer")
    pattern.register_server(server)
    
    # Create clients
    client1 = Client("Client1")
    client2 = Client("Client2")
    pattern.register_client(client1)
    pattern.register_client(client2)
    
    # Register clients with server
    server.register_client(client1)
    server.register_client(client2)
    
    # Send requests
    print("Sending requests...\n")
    
    response1 = pattern.route_request(
        "Client1", "MainServer", 
        RequestType.QUERY, 
        "Get user data for ID 123"
    )
    print(f"Response 1: {response1.status} (took {response1.processing_time:.2f}s)")
    
    response2 = pattern.route_request(
        "Client2", "MainServer",
        RequestType.COMMAND,
        "Update database record"
    )
    print(f"Response 2: {response2.status} (took {response2.processing_time:.2f}s)")
    
    response3 = pattern.route_request(
        "Client1", "MainServer",
        RequestType.QUERY,
        "Search for documents"
    )
    print(f"Response 3: {response3.status} (took {response3.processing_time:.2f}s)")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    
    print(f"\nServer Statistics:")
    for server_id, server_stats in stats["servers"].items():
        print(f"  {server_id}:")
        for key, value in server_stats.items():
            print(f"    {key}: {value}")
    
    print(f"\nClient Statistics:")
    for client_id, client_stats in stats["clients"].items():
        print(f"  {client_id}:")
        for key, value in client_stats.items():
            print(f"    {key}: {value}")

    

Agentic Architecture Pattern: Client Server

Pattern 122: 122 Client Server Pattern – Line-by-Line Explanation

Overview

Pattern 122: Client Server Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 122
• Pattern Name: 122 Client Server Pattern
• File Path: patterns\distributed_agent\122_client_server_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 122: Client Server Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 122 Client Server Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

ClientServerConfig (Line 18)

Description: Configuration for Client Server Pattern.

ClientServerState (Line 27)

Description: State information for Client Server Pattern.

ClientServerAgent (Line 36)

Description: Agent implementing Client Server Pattern.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Client Server Pattern."""

Purpose: Process input according to Client Server Pattern.

Parameters: – self: Instance reference – input_data: Parameter for input_data

get_statistics (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

ClientServerPattern (Line 85)

Description: Client Server Pattern – Enhanced implementation

This pattern provides comprehensive functionality for client server with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

Parameters: – self: Instance reference – agent: Parameter for agent

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

Parameters: – self: Instance reference – input_data: Parameter for input_data

get_statistics (Line 192)

def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = ClientServerConfig(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities