Skip to content

icecap.infrastructure.communication.rpc

Classes:

AgentClient

Bases: Protocol

High-level client for communicating with icecap-agent.

add_event_handler 

add_event_handler(callback: AgentClientEventHandler) -> None

Add an event handler for all received events.

Parameters:

Source code in icecap/infrastructure/communication/rpc/interface.py 
50
51
52
53
54
55
def add_event_handler(self, callback: AgentClientEventHandler) -> None:
    """Add an event handler for all received events.

    Args:
        callback: Function to call for each event
    """

close 

close() -> None

Close the connection to the agent.

Source code in icecap/infrastructure/communication/rpc/interface.py 
32
33
def close(self) -> None:
    """Close the connection to the agent."""

connect 

connect(timeout: float = 5.0) -> None

Connect to the agent.

Parameters:

  • timeout

    (float, default: 5.0 ) –

    Connection timeout in seconds

Raises:

Source code in icecap/infrastructure/communication/rpc/interface.py 
22
23
24
25
26
27
28
29
30
def connect(self, timeout: float = 5.0) -> None:
    """Connect to the agent.

    Args:
        timeout: Connection timeout in seconds

    Raises:
        AgentConnectionError: If connection fails
    """

is_connected 

is_connected() -> bool

Check if the client is connected to the agent.

Source code in icecap/infrastructure/communication/rpc/interface.py 
64
65
def is_connected(self) -> bool:
    """Check if the client is connected to the agent."""

remove_event_handler 

remove_event_handler(callback: AgentClientEventHandler) -> None

Remove an event handler.

Parameters:

Source code in icecap/infrastructure/communication/rpc/interface.py 
57
58
59
60
61
62
def remove_event_handler(self, callback: AgentClientEventHandler) -> None:
    """Remove an event handler.

    Args:
        callback: Handler to remove
    """

send 

send(command: Command, timeout: float = 5.0) -> Event

Send a command and wait for the response event.

Parameters:

  • command

    (Command) –

    Command protobuf to send

  • timeout

    (float, default: 5.0 ) –

    Maximum time to wait for response in seconds (None = wait forever)

Returns:

  • Event –

    The response event with matching operation_id

Raises:

Source code in icecap/infrastructure/communication/rpc/interface.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def send(self, command: commands_pb2.Command, timeout: float = 5.0) -> events_pb2.Event:
    """Send a command and wait for the response event.

    Args:
        command: Command protobuf to send
        timeout: Maximum time to wait for response in seconds (None = wait forever)

    Returns:
        The response event with matching operation_id

    Raises:
        AgentConnectionError: If not connected or connection fails
        AgentTimeoutError: If no response is received within timeout
    """

AgentClientEventHandler

Bases: Protocol

Callback protocol for handling agent events.

Can be either class-based or function-based.

__call__ 

__call__(event: Event) -> None

Handle an agent event.

Parameters:

  • event

    (Event) –

    The event received from the agent

Source code in icecap/infrastructure/communication/rpc/interface.py 
11
12
13
14
15
16
def __call__(self, event: events_pb2.Event) -> None:
    """Handle an agent event.

    Args:
        event: The event received from the agent
    """

AgentConnectionError

Bases: AgentError

Raised when connection to an agent fails or is lost.

AgentError

Bases: Exception

Base exception for agent client errors.

AgentTimeoutError

Bases: AgentError

Raised when waiting for a response times out.