tiphys.ipc.protocol
IPC Protocol definition for host-sandbox communication.
Defines the message format and types for the IPC mechanism.
1""" 2IPC Protocol definition for host-sandbox communication. 3 4Defines the message format and types for the IPC mechanism. 5""" 6 7from enum import Enum 8from typing import Any 9from uuid import uuid4 10 11from pydantic import BaseModel, Field 12 13 14class MessageType(str, Enum): 15 """Types of IPC messages.""" 16 17 REQUEST = "request" 18 RESPONSE = "response" 19 EVENT = "event" 20 ERROR = "error" 21 22 23class IPCMessage(BaseModel): 24 """Base IPC message structure.""" 25 26 id: str = Field(default_factory=lambda: str(uuid4())) 27 type: MessageType 28 channel: str = "default" # Routing channel (e.g. "sys", "agent") 29 payload: dict[str, Any] = Field(default_factory=dict) 30 metadata: dict[str, Any] = Field(default_factory=dict) 31 32 33class IPCRequest(IPCMessage): 34 """Request message expecting a response.""" 35 36 type: MessageType = MessageType.REQUEST 37 method: str 38 params: dict[str, Any] = Field(default_factory=dict) 39 40 41class IPCResponse(IPCMessage): 42 """Response to a request.""" 43 44 type: MessageType = MessageType.RESPONSE 45 request_id: str # ID of the request this responds to 46 result: Any = None 47 error: str | None = None 48 49 50class IPCEvent(IPCMessage): 51 """One-way event notification.""" 52 53 type: MessageType = MessageType.EVENT 54 event_name: str 55 data: Any = None
class
MessageType(builtins.str, enum.Enum):
15class MessageType(str, Enum): 16 """Types of IPC messages.""" 17 18 REQUEST = "request" 19 RESPONSE = "response" 20 EVENT = "event" 21 ERROR = "error"
Types of IPC messages.
REQUEST =
<MessageType.REQUEST: 'request'>
RESPONSE =
<MessageType.RESPONSE: 'response'>
EVENT =
<MessageType.EVENT: 'event'>
ERROR =
<MessageType.ERROR: 'error'>
class
IPCMessage(pydantic.main.BaseModel):
24class IPCMessage(BaseModel): 25 """Base IPC message structure.""" 26 27 id: str = Field(default_factory=lambda: str(uuid4())) 28 type: MessageType 29 channel: str = "default" # Routing channel (e.g. "sys", "agent") 30 payload: dict[str, Any] = Field(default_factory=dict) 31 metadata: dict[str, Any] = Field(default_factory=dict)
Base IPC message structure.
34class IPCRequest(IPCMessage): 35 """Request message expecting a response.""" 36 37 type: MessageType = MessageType.REQUEST 38 method: str 39 params: dict[str, Any] = Field(default_factory=dict)
Request message expecting a response.
Inherited Members
42class IPCResponse(IPCMessage): 43 """Response to a request.""" 44 45 type: MessageType = MessageType.RESPONSE 46 request_id: str # ID of the request this responds to 47 result: Any = None 48 error: str | None = None
Response to a request.
Inherited Members
51class IPCEvent(IPCMessage): 52 """One-way event notification.""" 53 54 type: MessageType = MessageType.EVENT 55 event_name: str 56 data: Any = None
One-way event notification.