KEI-Agent Python SDK

Enterprise-Grade Python SDK for KEI-Agent Framework with Multi-Protocol Support

The KEI-Agent Python SDK provides a unified, typed API for developing intelligent agents with comprehensive protocol support, enterprise security, and production monitoring.

🚀 Features

Multi-Protocol Support

• KEI-RPC: Synchronous request-response operations
• KEI-Stream: Bidirectional real-time communication
• KEI-Bus: Asynchronous message bus integration
• KEI-MCP: Model Context Protocol for tool integration

Enterprise Security

• Multi-Auth: Bearer Token, OIDC, mTLS
• Input Validation: Comprehensive sanitization and XSS/SQL injection protection
• Audit Logging: Complete traceability of all operations
• RBAC: Role-Based Access Control integration

Production Monitoring

• Structured Logging: JSON format with correlation IDs
• Health Checks: Database, API, memory, custom checks
• Performance Metrics: Built-in timing and resource monitoring
• Distributed Tracing: OpenTelemetry integration

Developer Experience

• Type Safety: 100% type hints for complete IntelliSense
• Comprehensive Documentation: Complete guides and API reference
• Auto-Protocol Selection: Intelligent protocol selection
• Async-First: Non-blocking I/O for maximum performance

📦 Installation

Installation from TestPyPI (Pre-Release)

pip install -i https://test.pypi.org/simple/ kei-agent-py-sdk

With extras (and fallback to PyPI for dependencies):

pip install -i https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  "kei-agent-py-sdk[security,docs]"

Standard Installation

pip install kei_agent_py_sdk

With Enterprise Features

pip install "kei_agent_py_sdk[security,docs]"

Development Installation

git clone https://github.com/oscharko-dev/kei-agent-py-sdk.git
cd kei-agent-py-sdk
pip install -e ".[dev,docs,security]"

⚡ Quick Start

Simple Agent Client

import asyncio
import psutil
import requests
from kei_agent import (
    UnifiedKeiAgentClient,
    AgentClientConfig,
    CapabilityManager,
    CapabilityProfile
)

# 1. TOOL IMPLEMENTATION: System Monitor
async def system_monitor_tool(target: str, metrics: list) -> dict:
    """Real implementation for system metrics using psutil."""
    result = {}

    if "cpu" in metrics:
        result["cpu_percent"] = psutil.cpu_percent(interval=1)
    if "memory" in metrics:
        memory = psutil.virtual_memory()
        result["memory_percent"] = memory.percent
    if "disk" in metrics:
        disk = psutil.disk_usage('/')
        result["disk_percent"] = (disk.used / disk.total) * 100

    return {
        "target": target,
        "metrics": result,
        "status": "healthy" if all(v < 80 for v in result.values()) else "warning"
    }

# 2. TOOL IMPLEMENTATION: API Health Check
async def api_health_tool(endpoint: str) -> dict:
    """Checks API endpoint availability."""
    try:
        response = requests.get(endpoint, timeout=5)
        return {
            "endpoint": endpoint,
            "status_code": response.status_code,
            "response_time_ms": response.elapsed.total_seconds() * 1000,
            "status": "healthy" if response.status_code == 200 else "unhealthy"
        }
    except Exception as e:
        return {
            "endpoint": endpoint,
            "error": str(e),
            "status": "unhealthy"
        }

async def main():
    config = AgentClientConfig(
        base_url="https://api.kei-framework.com",
        api_token="your-api-token",
        agent_id="my-agent"
    )

    async with UnifiedKeiAgentClient(config=config) as client:
        # 3. REGISTER TOOLS
        capability_manager = CapabilityManager(client._legacy_client)

        # Register system monitor tool
        await capability_manager.register_capability(
            CapabilityProfile(
                name="system_monitor",
                version="1.0.0",
                description="Collects CPU, memory, disk metrics",
                methods={"get_metrics": {"parameters": ["target", "metrics"]}}
            ),
            handler=system_monitor_tool
        )

        # Register API health tool
        await capability_manager.register_capability(
            CapabilityProfile(
                name="api_health_checker",
                version="1.0.0",
                description="Checks API endpoint availability",
                methods={"check_endpoint": {"parameters": ["endpoint"]}}
            ),
            handler=api_health_tool
        )

        # 4. USE COMPLETE IMPLEMENTATION
        # Plan with concrete tools
        plan = await client.plan_task(
            objective="Perform complete system diagnosis",
            context={"tools": ["system_monitor", "api_health_checker"]}
        )
        print(f"Plan created: {plan['plan_id']}")

        # Get system metrics via registered tool
        system_data = await client.use_tool(
            "system_monitor",
            **{
                "target": "localhost",
                "metrics": ["cpu", "memory", "disk"]
            }
        )
        print(f"System metrics: {system_data}")

        # Check API health via registered tool
        api_data = await client.use_tool(
            "api_health_checker",
            **{"endpoint": "https://api.kei-framework.com/health"}
        )
        print(f"API status: {api_data['status']}")

asyncio.run(main())

Multi-Protocol Features

import asyncio
import time
from kei_agent import ProtocolType

async def multi_protocol_example():
    config = AgentClientConfig(
        base_url="https://api.kei-framework.com",
        api_token="your-api-token",
        agent_id="multi-protocol-agent"
    )

    async with UnifiedKeiAgentClient(config=config) as client:
        # Automatic protocol selection (RPC) - Correct API signature
        plan = await client.plan_task(
            objective="Discover available tools",
            context={"category": "monitoring", "max_results": 5}
        )
        print(f"Plan: {plan}")

        # Streaming: Use execute_agent_operation for stream operations
        stream_result = await client.execute_agent_operation(
            "stream_monitoring",
            {"data": "real-time-feed", "callback": True},
            protocol=ProtocolType.STREAM
        )
        print(f"Stream result: {stream_result}")

        # Tool discovery via MCP - Concrete implementable tools
        tools = await client.discover_available_tools("monitoring")
        print(f"Available tools: {len(tools)}")

        # Use available tool (if present)
        if tools:
            tool_result = await client.use_tool(
                tools[0]["name"],
                **{"target": "system", "check_type": "basic"}
            )
            print(f"Tool result: {tool_result}")

        # Asynchronous bus operation - Concrete implementation
        bus_result = await client.execute_agent_operation(
            "async_health_check",
            {
                "target_agent": "monitoring-agent",
                "message_type": "health_check_request",
                "payload": {"scope": "basic", "timeout": 30}
            },
            protocol=ProtocolType.BUS
        )
        print(f"Bus result: {bus_result}")

asyncio.run(multi_protocol_example())

Enterprise Features

import time
from kei_agent import (
    get_logger,
    get_health_manager,
    LogContext,
    APIHealthCheck,
    MemoryHealthCheck,
    HealthStatus
)

# Structured Logging
logger = get_logger("enterprise_agent")
# create_correlation_id() already sets the context
correlation_id = logger.create_correlation_id()
logger.set_context(LogContext(
    user_id="user-123",
    agent_id="enterprise-agent"
))

# Health Monitoring
health_manager = get_health_manager()
health_manager.register_check(APIHealthCheck(
    name="external_api",
    url="https://api.external.com/health"
))
health_manager.register_check(MemoryHealthCheck(
    name="system_memory",
    warning_threshold=0.8
))

async def enterprise_example():
    config = AgentClientConfig(
        base_url="https://api.kei-framework.com",
        api_token="your-api-token",
        agent_id="enterprise-agent"
    )

    async with UnifiedKeiAgentClient(config=config) as client:
        # Operation with logging
        operation_id = logger.log_operation_start("business_process")
        start_time = time.time()

        try:
            result = await client.plan_task("Enterprise task")
            logger.log_operation_end("business_process", operation_id, start_time, success=True)

            # Health Check
            summary = await health_manager.run_all_checks()
            logger.info(
                "Health check completed",
                overall_status=summary.overall_status.value,
                healthy_count=summary.healthy_count,
            )

        except Exception as e:
            logger.log_operation_end("business_process", operation_id, start_time, success=False)
            logger.error("Business process failed", error=str(e))
            raise

asyncio.run(enterprise_example())

🏗️ Architecture

The SDK follows a modular, enterprise-grade architecture:

kei_agent/
├── unified_client.py               # Main API class
├── protocol_types.py               # Type definitions and configurations
├── security_manager.py             # Authentication and token management
├── protocol_clients.py             # KEI-RPC, Stream, Bus, MCP clients
├── protocol_selector.py            # Intelligent protocol selection
├── enterprise_logging.py           # Structured JSON logging
├── health_checks.py               # System monitoring and health checks
└── input_validation.py            # Input validation and sanitization

Design Principles

• Clean Code: All modules ≤200 lines, functions ≤20 lines
• Type Safety: 100% type hints for all public APIs
• Single Responsibility: Each module has a clearly defined responsibility
• Async-First: Non-blocking I/O for maximum performance
• Enterprise-Ready: Production monitoring and security hardening

📚 Documentation

• Complete Documentation - Comprehensive guides and API reference

🔧 Configuration

Basic Configuration

from kei_agent import AgentClientConfig, ProtocolConfig, SecurityConfig, AuthType

# Agent configuration
agent_config = AgentClientConfig(
    base_url="https://api.kei-framework.com",
    api_token="your-api-token",
    agent_id="my-agent",
    timeout=30,
    max_retries=3
)

# Protocol configuration
protocol_config = ProtocolConfig(
    rpc_enabled=True,
    stream_enabled=True,
    bus_enabled=True,
    mcp_enabled=True,
    auto_protocol_selection=True,
    protocol_fallback_enabled=True
)

# Security configuration
security_config = SecurityConfig(
    auth_type=AuthType.BEARER,
    api_token="your-api-token",
    rbac_enabled=True,
    audit_enabled=True
)

# Client with complete configuration
client = UnifiedKeiAgentClient(
    config=agent_config,
    protocol_config=protocol_config,
    security_config=security_config
)

Environment Variables

export KEI_API_URL="https://api.kei-framework.com"
export KEI_API_TOKEN="your-api-token"
export KEI_AGENT_ID="my-agent"
export KEI_AUTH_TYPE="bearer"
export KEI_RBAC_ENABLED="true"
export KEI_AUDIT_ENABLED="true"

🧪 Testing

# Run unit tests
python -m pytest tests/ -v

# With coverage
python -m pytest tests/ --cov=kei_agent --cov-report=html

# Specific test categories
python -m pytest tests/ -m "unit"          # Unit tests
python -m pytest tests/ -m "integration"   # Integration tests
python -m pytest tests/ -m "security"      # Security tests

# Performance tests
python -m pytest tests/ -m "performance"

🤝 Contributing

We welcome contributions! Please read our Development Guide and Contribution Guidelines.

Development Setup

# Clone repository
git clone https://github.com/oscharko-dev/kei-agent-py-sdk.git
cd kei-agent-py-sdk

# Set up development environment
python -m venv venv
source venv/bin/activate  # Linux/macOS
pip install -e ".[dev,docs,security]"

# Install pre-commit hooks
pre-commit install

# Run tests
make test

# Build documentation
mkdocs build --strict

📄 License

This project is licensed under the MIT License.

🔗 Links

• GitHub Repository: oscharko-dev/kei-agent-py-sdk
• TestPyPI Package: kei-agent-py-sdk
• Documentation: GitHub Pages
• Issues: GitHub Issues

📊 Status

• ✅ Production Ready: Fully tested and documented
• ✅ Type Safe: 100% type hints for all APIs
• ✅ Enterprise Grade: Security, monitoring, and compliance features
• ✅ Well Documented: Comprehensive documentation
• ✅ Actively Maintained: Regular updates and support

Ready to get started? Install the SDK and follow our Quick Start Guide!