What is ACP ?

"""
ACP Server - SreeniParrotAgent Implementation

This server implements a simple ACP (Agent Communication Protocol) server with a single agent
that echoes back received messages after a 30-second processing delay.

Dependencies:
- acp_sdk: The ACP SDK for building agent servers
- asyncio: For asynchronous operations

Usage:
    python main.py

The server will start on http://0.0.0.0:8000
"""

import asyncio
from acp_sdk.server import Server
from typing import AsyncGenerator

from acp_sdk import MessagePart

# Initialize the ACP server instance
server = Server()

@server.agent(
    name="SreeniParrotAgent",  # Unique identifier for the agent
    description="An agent that echoes back the received message.",  # Human-readable description
    metadata={
        "version": "1.0",  # Agent version
        "author": {
            "name": "Sreeni",  # Author name
            "email": "sreeni@example.com"  # Author contact
        }
    }
)
async def SreeniParrotAgent(messages: list) -> AsyncGenerator[MessagePart, None]:
    """
    Main agent function that processes incoming messages.

    Args:
        messages (list): List of message objects containing input data

    Yields:
        MessagePart: Processed output messages

    Processing Flow:
        1. Receives messages from client
        2. Waits 30 seconds (simulating processing time)
        3. Echoes back the original content
    """
    for message in messages:
        # Process each message in the input
        for part in message.parts:
            # Simulate processing delay (30 seconds)
            await asyncio.sleep(30)

            # Echo back the received content as output
            yield MessagePart(
                name="output",           # Output identifier
                content_type="text",     # Content type (text in this case)
                content=part.content     # The actual content to echo back
            )

if __name__ == "__main__":
    # Start the server when script is run directly
    server.run(host="0.0.0.0", port=8000)

"""
ACP Client - Endpoint Testing Suite

This client tests all available endpoints of the ACP server and displays
comprehensive status information and response content.

Dependencies:
- requests: HTTP library for making API calls

Usage:
    python client.py

Prerequisites:
    - ACP server must be running on http://0.0.0.0:8000
"""

import requests
import json

# Base URL for the ACP server
BASE_URL = "http://0.0.0.0:8000"

# ANSI color codes for terminal output
class Colors:
    BLUE = '33[94m'      # Blue for URLs
    GREEN = '33[92m'     # Green for success
    YELLOW = '33[93m'    # Yellow for warnings
    RED = '33[91m'       # Red for errors
    PURPLE = '33[95m'    # Purple for info
    CYAN = '33[96m'      # Cyan for status
    BOLD = '33[1m'       # Bold text
    UNDERLINE = '33[4m'  # Underlined text
    END = '33[0m'        # Reset color

# Example payload for creating a new run
# This payload tells the server to process "Hello, World!" through SreeniParrotAgent
example_payload = {
    "agent_name": "SreeniParrotAgent",  # Which agent to use
    "input": [
        {
            "parts": [
                {
                    "name": "input",           # Input identifier
                    "content_type": "text",    # Type of content
                    "content": "Hello, World!" # Actual message content
                }
            ]
        }
    ]
}

# Payload for resuming an existing run
# This allows adding more messages to an ongoing conversation
resume_payload = {
    "await_resume": {
        "enabled": True,  # Enable resume functionality
        "message": {
            "parts": [
                {
                    "name": "input",
                    "content_type": "text",
                    "content": "Resuming run"
                }
            ]
        }
    },
    "mode": "sync",  # Synchronous processing mode
    "messages": [
        {
            "parts": [
                {
                    "name": "input",
                    "content_type": "text",
                    "content": "Hello again!"  # Additional message content
                }
            ]
        }
    ]
}

def print_detailed_status(data, endpoint_name):
    """
    Display comprehensive status information from API responses.

    Args:
        data: Response data from the API
        endpoint_name (str): Name of the endpoint being tested
    """
    print(f"n{Colors.BOLD}{endpoint_name} - Detailed Status:{Colors.END}")
    print("-" * 50)

    if isinstance(data, dict):
        for key, value in data.items():
            # Handle status information specially
            if key == "status_info" and isinstance(value, dict):
                print(f"  {Colors.CYAN}STATUS INFORMATION:{Colors.END}")
                for status_key, status_value in value.items():
                    print(f"    {status_key}: {status_value}")

            # Handle agent lists specially
            elif key == "agents" and isinstance(value, list):
                print(f"  🤖 {Colors.PURPLE}{key.upper()}:{Colors.END}")
                for agent in value:
                    print(f"    Name: {agent.get('name')}")
                    print(f"    Status: {agent.get('metadata', {}).get('status', 'N/A')}")
                    print(f"    Version: {agent.get('metadata', {}).get('version', 'N/A')}")

            # Handle output content specially
            elif key == "output" and isinstance(value, list):
                print(f"  �� {Colors.GREEN}{key.upper()}:{Colors.END}")
                for i, output in enumerate(value):
                    print(f"    Output {i+1}: {output.get('content', 'N/A')}")
                    if 'metadata' in output and 'status_info' in output['metadata']:
                        print(f"      Status: {output['metadata']['status_info']}")

            # Display other data normally
            else:
                print(f"  {Colors.YELLOW}{key.upper()}:{Colors.END} {value}")
    else:
        print(f"  Content: {data}")

# ============================================================================
# ENDPOINT FUNCTIONS
# ============================================================================

def list_agents():
    """
    GET /agents - Retrieve list of all available agents
    """
    endpoint_url = f"{BASE_URL}/agents"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
    response = requests.get(endpoint_url)
    print(f"List Agents: {Colors.CYAN}{response.status_code}{Colors.END}")
    if response.status_code == 200:
        data = response.json()
        print_detailed_status(data, "List Agents")

def read_agent(name):
    """
    GET /agents/{name} - Get detailed information about a specific agent

    Args:
        name (str): Name of the agent to retrieve
    """
    endpoint_url = f"{BASE_URL}/agents/{name}"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
    response = requests.get(endpoint_url)
    print(f"Read Agent {name}: {Colors.CYAN}{response.status_code}{Colors.END}")
    if response.status_code == 200:
        data = response.json()
        print_detailed_status(data, f"Read Agent {name}")

def ping():
    """
    GET /ping - Health check endpoint to verify server is running
    """
    endpoint_url = f"{BASE_URL}/ping"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
    response = requests.get(endpoint_url)
    print(f"Ping: {Colors.CYAN}{response.status_code}{Colors.END}")
    data = response.json()
    print_detailed_status(data, "Ping")

def create_run(payload):
    """
    POST /runs - Create a new run with the specified agent and input

    Args:
        payload (dict): Run configuration including agent name and input data

    Returns:
        dict: Run response data if successful, None otherwise
    """
    endpoint_url = f"{BASE_URL}/runs"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}POST {endpoint_url}{Colors.END}")
    response = requests.post(endpoint_url, json=payload)
    print(f"Create Run: {Colors.CYAN}{response.status_code}{Colors.END}")
    if response.status_code == 200:
        data = response.json()
        print_detailed_status(data, "Create Run")
        return data
    return None

def read_run(run_id):
    """
    GET /runs/{run_id} - Get status and output of a specific run

    Args:
        run_id (str): Unique identifier of the run to retrieve
    """
    endpoint_url = f"{BASE_URL}/runs/{run_id}"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
    response = requests.get(endpoint_url)
    print(f"Read Run {run_id}: {Colors.CYAN}{response.status_code}{Colors.END}")
    if response.status_code == 200:
        data = response.json()
        print_detailed_status(data, f"Read Run {run_id}")

def resume_run(run_id, payload):
    """
    POST /runs/{run_id} - Resume an existing run with additional input

    Args:
        run_id (str): Unique identifier of the run to resume
        payload (dict): Additional input data for the run
    """
    endpoint_url = f"{BASE_URL}/runs/{run_id}"
    print(f"n�� Testing: {Colors.BLUE}{Colors.BOLD}POST {endpoint_url}{Colors.END}")
    response = requests.post(endpoint_url, json=payload)
    print(f"Resume Run {run_id}: {Colors.CYAN}{response.status_code}{Colors.END}")
    if response.status_code == 200:
        data = response.json()
        print_detailed_status(data, f"Resume Run {run_id}")
    else:
        print(f"{Colors.RED}Error: {response.text}{Colors.END}")



# ============================================================================
# MAIN EXECUTION
# ============================================================================

if __name__ == "__main__":
    print(f"{Colors.BOLD}Testing ACP Endpoints with Full Status Display{Colors.END}")
    print("=" * 60)

    # Test basic server functionality
    print(f"n{Colors.PURPLE}📡 Testing Basic Server Endpoints...{Colors.END}")
    ping()



    # Test agent management endpoints
    print(f"n{Colors.PURPLE}🤖 Testing Agent Endpoints...{Colors.END}")
    list_agents()
    read_agent("SreeniParrotAgent")

    # Test run management endpoints
    print(f"n{Colors.PURPLE}🔄 Testing Run Endpoints...{Colors.END}")

    # Create a new run and get the run ID for further testing
    print(f"n{Colors.PURPLE} Creating a new run to test run-specific endpoints...{Colors.END}")
    run_result = create_run(example_payload)

    if run_result and 'run_id' in run_result:
        run_id = run_result['run_id']

        # Test reading the created run
        read_run(run_id)

        # Test resuming the run with additional input
        resume_run(run_id, resume_payload)
    else:
        print(f"{Colors.RED}❌ Failed to create run - cannot test run-specific endpoints{Colors.END}")