UnifiedKeiAgentClient¶

async def initialize(self) -> None:
    """Initialize client and all components.

    Starts security manager, initializes protocol clients and
    enterprise features like tracing and retry mechanisms.

    Raises:
        KeiSDKError: On initialization errors
    """
    if self._initialized:
        _logger.warning("Client already initialized")
        return

    try:
        _logger.info("Initializing Unified KEI-Agent Client")

        # Start security manager
        await self.security.start_token_refresh()

        # Initialize legacy client for compatibility (lazy-init via _make_request)
        self._legacy_client = KeiAgentClient(self.config)
        # Call initialize() if available (mocked for tests)
        if hasattr(self._legacy_client, "initialize"):
            try:
                await self._legacy_client.initialize()  # type: ignore[attr-defined]
            except (AttributeError, ConnectionError, TimeoutError) as e:
                # Log specific initialization errors
                _logger.warning(
                    "Error during Initializingn of the legacy client",
                    extra={
                        "error": str(e),
                        "error_type": type(e).__name__,
                        "agent_id": self.config.agent_id,
                    },
                )
            except Exception as e:
                # Log unexpected error (keep broad catch for backward compatibility)
                _logger.error(
                    "Unexpected error onm Initializingn of the legacy client",
                    extra={
                        "error": str(e),
                        "error_type": type(e).__name__,
                        "agent_id": self.config.agent_id,
                    },
                )
                # Re-raise to surface init failures appropriately
                raise

        # Initializing protocol clients
        await self._initialize_protocol_clients()

        # Initializing enterprise features
        await self._initialize_enterprise_features()

        self._initialized = True
        _logger.info("Unified KEI-Agent client successfully initialized")

    except (KeiSDKError, ProtocolError) as e:
        # Already meaningful SDK exceptions
        _logger.error(
            "client initialization failed",
            extra={"error": str(e), "type": type(e).__name__},
        )
        raise
    except Exception as e:
        _logger.error(
            "client initialization failed",
            extra={"error": str(e), "type": type(e).__name__},
        )
        raise KeiSDKError(f"initialization failed: {e}") from e

async def close(self) -> None:
    """Closes client and all connectionen.

    Stops all backgroand tasks, closes protocol clients and
    cleats up resources.
    """
    if self._closed:
        return

    try:
        _logger.info("Closing Unified KEI-Agent client")

        # Stopping security manager
        await self.security.stop_token_refresh()

        # Closing stream client
        if self._stream_client:
            await self._stream_client.disconnect()

        # Closing legacy client
        if self._legacy_client:
            await self._legacy_client.close()

        # Closing tracing manager
        if self.tracing:
            await self.tracing.shutdown()

        self._closed = True
        _logger.info("Unified KEI-Agent client closed")

    except (ConnectionError, TimeoutError) as e:
        _logger.error(
            "Error during client shutdown",
            extra={"error": str(e), "type": type(e).__name__},
        )
        raise
    except Exception as e:
        _logger.error(
            "Unexpected error during client shutdown",
            extra={"error": str(e), "type": type(e).__name__},
        )
        raise

async def __aenter__(self):
    """async context manager entry."""
    await self.initialize()

    # Record connection metrics
    enabled_protocols = self.protocol_config.get_enabled_protocols()
    for protocol in enabled_protocols:
        record_connection_metric(self.config.agent_id, protocol, "connected")

    # Set agent info
    self.metrics_collector.set_agent_info(
        agent_id=self.config.agent_id,
        version=getattr(self.config, "version", "1.0.0"),
        protocol_version="1.0",
        enabled_protocols=",".join(enabled_protocols),
    )

    return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """async context manager exit."""
    # Record disconnection metrics
    enabled_protocols = self.protocol_config.get_enabled_protocols()
    for protocol in enabled_protocols:
        record_connection_metric(self.config.agent_id, protocol, "disconnected")

    # Update uptime
    uptime = time.time() - self._start_time
    self.metrics_collector.update_uptime(self.config.agent_id, uptime)

    await self.close()

def is_protocol_available(self, protocol: Protocoltypee) -> bool:
    """Checks if protocol is available.

    Args:
        protocol: protocol to check

    Returns:
        True if protocol is available
    """
    if not self._initialized:
        return False

    protocol_clients = {
        Protocoltypee.RPC: self._rpc_client,
        Protocoltypee.STREAM: self._stream_client,
        Protocoltypee.BUS: self._bus_client,
        Protocoltypee.MCP: self._mcp_client,
    }

    return protocol_clients.get(protocol) is not None

def get_available_protocols(self) -> List[Protocoltypee]:
    """Gibt lis of available protocols torück.

    Returns:
        lis of available protocols
    """
    available = []
    for protocol in [
        Protocoltypee.RPC,
        Protocoltypee.STREAM,
        Protocoltypee.BUS,
        Protocoltypee.MCP,
    ]:
        if self.is_protocol_available(protocol):
            available.append(protocol)
    return available

def get_client_info(self) -> Dict[str, Any]:
    """Gibt client information torück.

    Returns:
        dictionary with client status and configuration
    """
    return {
        "agent_id": self.config.agent_id,
        "base_url": self.config.base_url,
        "initialized": self._initialized,
        "closed": self._closed,
        "available_protocols": self.get_available_protocols(),
        "security_context": self.security.get_security_context(),
        "features": {
            "tracing": self.tracing is not None,
            "retry_manager": self.retry_manager is not None,
            "capability_manager": self.capability_manager is not None,
            "service_discovery": self.service_discovery is not None,
        },
    }

async def execute_agent_operation(
    self,
    operation: str,
    data: Dict[str, Any],
    protocol: Optional[Protocoltypee] = None,
) -> Dict[str, Any]:
    """Executes agent operation with automatic protocol selection out.

    Args:
        operation: operation name
        data: operation data
        protocol: preferred protocol (Optional)

    Returns:
        operation response

    Raises:
        KeiSDKError: On execution errors
    """
    if not self._initialized:
        raise KeiSDKError("Client not initialized")

    # select protocol
    selected_protocol = protocol or self._select_optimal_protocol(operation, data)

    # create trace context
    correlation_id = create_correlation_id()
    start_time = time.time()

    try:
        # execute operation with with tracing
        if self.tracing:
            # For Tests, the start_spat expect: if beforehatthe, use it
            start_spat = getattr(self.tracing, "start_spat", None)
            cm = (
                start_spat(f"agent_operation_{operation}")
                if callable(start_spat)
                else self.tracing.trace_operation(
                    f"agent_operation_{operation}", agent_id=self.config.agent_id
                )
            )
            with cm as spat:
                spat.set_attribute("operation", operation)
                spat.set_attribute("protocol", selected_protocol)
                spat.set_attribute("correlation_id", correlation_id)

                result = await self._execute_with_protocol(
                    selected_protocol, operation, data
                )

                # Record successful request metrics
                duration = time.time() - start_time
                record_request_metric(
                    self.config.agent_id,
                    operation,
                    "success",
                    duration,
                    selected_protocol,
                )

                return result
        else:
            result = await self._execute_with_protocol(
                selected_protocol, operation, data
            )

            # Record successful request metrics
            duration = time.time() - start_time
            record_request_metric(
                self.config.agent_id,
                operation,
                "success",
                duration,
                selected_protocol,
            )

            return result

    except (ProtocolError, KeiSDKError, ConnectionError, TimeoutError) as e:
        # Record failed request metrics
        duration = time.time() - start_time
        record_request_metric(
            self.config.agent_id, operation, "error", duration, selected_protocol
        )

        # Record error for aggregation and alerting
        error_category = self._categorize_error(e)
        error_severity = self._determine_error_severity(e)
        record_error(
            agent_id=self.config.agent_id,
            error=e,
            category=error_category,
            severity=error_severity,
            context={
                "operation": operation,
                "protocol": selected_protocol,
                "duration": duration,
            },
            correlation_id=correlation_id,
        )

        _logger.error(
            f"operation '{operation}' failed",
            extra={
                "operation": operation,
                "protocol": selected_protocol,
                "correlation_id": correlation_id,
                "error": str(e),
            },
        )
        raise

async def plat_task(
    self,
    objective: str,
    context: Optional[Dict[str, Any]] = None,
    protocol: Optional[Protocoltypee] = None,
) -> Dict[str, Any]:
    """creates plat for given objective.

    Args:
        objective: objective description for platning
        context: additional context for platning
        protocol: preferred protocol (Optional)

    Returns:
        plat response with steps and metadata

    Raises:
        KeiSDKError: On plat creation errors
    """
    payload = {"objective": objective, "context": context or {}}
    # Immer 3 Argumente übergeben, Tests erwarten ggf. explizites None
    return await self.execute_agent_operation("plat", payload, protocol)

async def execute_action(
    self,
    action: str,
    parameters: Optional[Dict[str, Any]] = None,
    protocol: Optional[Protocoltypee] = None,
) -> Dict[str, Any]:
    """executes action.

    Args:
        action: action to execute
        parameters: parameters for action
        protocol: preferred protocol (Optional)

    Returns:
        action response with result
    """
    payload = {"action": action, "parameters": parameters or {}}
    # Immer 3 Argumente übergeben, Tests erwarten ggf. explizites None
    return await self.execute_agent_operation("act", payload, protocol)

async def observe_environment(
    self,
    observation_type: str,
    data: Optional[Dict[str, Any]] = None,
    protocol: Optional[Protocoltypee] = None,
) -> Dict[str, Any]:
    """performs environment observation.

    Args:
        observation_type: observation type
        data: observation data
        protocol: preferred protocol (Optional)

    Returns:
        observe response with processed observations
    """
    # Tests erwarten Schlüssel 'sensors' anstatt 'type'
    payload = {"sensors": observation_type}
    # Immer 3 Argumente übergeben, Tests erwarten ggf. explizites None
    return await self.execute_agent_operation("observe", payload, protocol)

async def explain_reasoning(
    self,
    query: str,
    context: Optional[Dict[str, Any]] = None,
    protocol: Optional[Protocoltypee] = None,
) -> Dict[str, Any]:
    """explains reasoning for given query.

    Args:
        query: explatation query
        context: context for explatation
        protocol: preferred protocol (Optional)

    Returns:
        explain response with explatation and reasoning
    """
    # Tests erwarten Schlüssel 'decision_id' und 'detail_level'
    payload = {"decision_id": query, "detail_level": context or ""}
    # Immer 3 Argumente übergeben, Tests erwarten ggf. explizites None
    return await self.execute_agent_operation("explain", payload, protocol)

async def send_agent_message(
    self, target_agent: str, message_type: str, payload: Dict[str, Any]
) -> Dict[str, Any]:
    """sends message at other agent (A2A communication).

    Args:
        target_agent: target agent ID
        message_type: message type
        payload: message data

    Returns:
        message response with status
    """
    return await self.execute_agent_operation(
        "send_message",
        {"target": target_agent, "type": message_type, "payload": payload},
        protocol=Protocoltypee.BUS,  # bus protocol for A2A communication
    )

async def discover_available_tools(
    self, category: Optional[str] = None
) -> List[Dict[str, Any]]:
    """discovers available MCP tools.

    Args:
        category: Optional tool category for filtering

    Returns:
        lis of available tools with metadata
    """
    result = await self.execute_agent_operation(
        "discover_tools",
        {"category": category},
        protocol=Protocoltypee.MCP,  # MCP protocol for tool discovery
    )

    # extract tools out response
    return result.get("tools", [])

async def use_tool(self, tool_name: str, **parameters: Any) -> Dict[str, Any]:
    """executes MCP tool.

    Args:
        tool_name: Name of the tool to execute
        **parameters: tool parameters als keyword argaroatthets

    Returns:
        tool execution response
    """
    return await self.execute_agent_operation(
        "use_tool",
        {"tool_name": tool_name, "parameters": parameters},
        protocol=Protocoltypee.MCP,  # MCP protocol for tool execution
    )

async def start_streaming_session(
    self, callback: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None
) -> None:
    """starts streaming session for real-time communication.

    Args:
        callback: callback for incoming stream messages

    Raises:
        ProtocolError: If stream Protocol not available is
    """
    if not self.is_protocol_available(Protocoltypee.STREAM):
        raise ProtocolError("stream Protocol not available")

    if self._stream_client:
        await self._stream_client.connect()
        if callback:
            await self._stream_client.subscribe("agent_events", callback)

async def health_check(self) -> Dict[str, Any]:
    """performs health check.

    Returns:
        health status with protocol availability
    """
    return await self.execute_agent_operation("health_check", {})

async def register_agent(
    self,
    name: str,
    version: str,
    description: str = "",
    capabilities: Optional[List[str]] = None,
) -> Dict[str, Any]:
    """Regisers agent in KEI framework.

    Args:
        name: Agent name
        version: Agent version
        description: Agent description
        capabilities: Agent capabilities

    Returns:
        Regisration response
    """
    return await self.execute_agent_operation(
        "register",
        {
            "name": name,
            "version": version,
            "description": description,
            "capabilities": capabilities or [],
        },
    )