Kimi CLI Wire Message Types in `src/kimi_cli/wire/types.py`: Events, Requests, and Envelope Protocol

src/kimi_cli/wire/types.py defines every Wire message type that the Kimi CLI uses to exchange information between the core "soul" runtime and any UI or client, grouping them into Event and Request unions that are transmitted inside a WireMessageEnvelope.

The MoonshotAI/kimi-cli repository implements a structured JSON-line protocol to keep agent logic decoupled from interface code. All canonical message variants, helper functions, and envelope logic live in src/kimi_cli/wire/types.py and are exported through the module's __all__ list (lines 64–78). If you are building a custom client or debugging message flow, these Wire message types are the authoritative contract for the wire protocol.

Defined Wire Message Categories

The file organizes concrete message variants into two disjoint categories: Events that fire unidirectionally, and Requests that block until the client provides a reply.

Event Types: Control Flow, Status, and UI Payloads

Events represent something that happened inside the runtime. According to the source code, the Event union contains the following concrete types:

• TurnBegin – Signals the start of a new user turn.
• SteerInput – Carries mid-turn steering or guidance from the user.
• TurnEnd – Marks the completion of a turn.
• StepBegin – Fires when a new reasoning or action step starts.
• StepInterrupted – Emitted if a step is cut short.
• StepRetry – Indicates the runtime is retrying a failed step.
• CompactionBegin / CompactionEnd – Bookend a memory compaction operation.
• MCPLoadingBegin / MCPLoadingEnd – Notify the client that MCP resources are loading.
• StatusUpdate – Carries general runtime status text.
• MCPServerSnapshot / MCPStatusSnapshot – Deliver point-in-time MCP server and status data.
• Notification – General notice to the client.
• ContentPart – A fragment of streamed model content.
• ToolCall / ToolCallPart – Represent an invoked tool or a partial tool payload.
• ToolResult – Returns the output of a tool execution.
• ApprovalResponse – Conveys the result of an approval check.
• SubagentEvent – Relays activity from a sub-agent.
• PlanDisplay – Renders a plan or strategy to the UI.
• BtwBegin / BtwEnd – Frame a side question or "by the way" interaction.

These variants compose the Event union type declared in the module.

Request Types: Client Interaction Prompts

Requests are sent from the runtime to the client and expect a user- or client-generated response. The concrete request models defined in src/kimi_cli/wire/types.py include:

• ApprovalRequest – Asks the user to approve a sensitive action.
• ToolCallRequest – Requests permission or capability to execute a specific tool.
• QuestionOption / QuestionItem / QuestionResponse – Support structured question answering.
• QuestionRequest – Presents a question to the user and awaits an answer.
• QuestionNotSupported – Indicates the runtime cannot handle a given question type.

The Request union is built as type Request = (ApprovalRequest | ToolCallRequest | QuestionRequest | HookRequest). All request symbols appear alongside the Event types in the module-level __all__ definition, making them the canonical WireMessage variants available for import. Downstream code typically consumes these through kimi_cli.wire, which re-exports the public API so other packages do not depend on the internal module layout.

Union Types and Type Hierarchy

At the top level, the module exposes three composable union aliases:

type Event = (TurnBegin | SteerInput | … | BtwEnd)
type Request = (ApprovalRequest | ToolCallRequest | QuestionRequest | HookRequest)
type WireMessage = Event | Request

Every concrete model ultimately resolves to WireMessage, which is the broadest type used when the runtime or client does not yet know which variant was received.

WireMessageEnvelope and JSON-Line Protocol

Raw messages are never transmitted directly. Instead, they are wrapped in a WireMessageEnvelope (lines 34–62) that adds a "type" discriminator field. This envelope lets the receiver reconstruct the exact Pydantic model from a single JSON line.

The envelope exposes two critical methods:

• WireMessageEnvelope.from_wire_message(msg) – Wraps a concrete WireMessage instance before serialization.
• WireMessageEnvelope.to_wire_message() – Parses the envelope's "type" field and returns the matching concrete model.

Because the protocol operates over newline-delimited JSON, each envelope.model_dump_json() call produces one line that the client reads and reconstructs with WireMessageEnvelope.model_validate_json(line).

Runtime Type Checking Helpers

To avoid manual isinstance chains, the module provides three fast predicate functions at lines 12–25:

• is_event(msg) – Returns True if the message is an instance of the Event union.
• is_request(msg) – Returns True if the message is an instance of the Request union.
• is_wire_message(msg) – Returns True for any member of the WireMessage union.

These helpers are the preferred way to route messages inside UI loops and runtime dispatch code.

Practical Code Examples

Sending a TurnBegin Event

from kimi_cli.wire.types import TurnBegin, WireMessageEnvelope

# Build the event

event = TurnBegin(user_input="Explain the plan for the next step.")

# Wrap it for transmission

envelope = WireMessageEnvelope.from_wire_message(event)

# Serialize to JSON (the UI/client reads a line at a time)

json_line = envelope.model_dump_json()
print(json_line)   # → {"type":"TurnBegin","payload":{"user_input":"Explain the plan for the next step."}}

Receiving and Routing Messages by Type

from kimi_cli.wire.types import WireMessageEnvelope, is_event, is_request

# Imagine `line` is a JSON string read from the client

envelope = WireMessageEnvelope.model_validate_json(line)
msg = envelope.to_wire_message()          # Re‑creates the original Pydantic model

if is_event(msg):
    print("Got an event:", type(msg).__name__)   # e.g. TurnBegin

elif is_request(msg):
    print("Got a request:", type(msg).__name__)  # e.g. ToolCallRequest

Handling a ToolCallRequest and Responding with ToolResult

from kimi_cli.wire.types import ToolCallRequest, ToolResult, WireMessageEnvelope

# Received request (already deserialized)

request: ToolCallRequest = ...

# Execute the tool (pseudo‑code)

result = execute_tool(request.name, request.arguments)

# Build the result message

tool_result = ToolResult(
    tool_call_id=request.id,
    result=result,
    display=[],
)

# Send back to client

envelope = WireMessageEnvelope.from_wire_message(tool_result)
send_to_client(envelope.model_dump_json())

Summary

• src/kimi_cli/wire/types.py is the source of truth for every Wire message type in the Kimi CLI.
• Messages are split into Events (unidirectional) and Requests (awaiting reply), which together form the WireMessage union.
• The WireMessageEnvelope handles serialization and deserialization over newline-delimited JSON using a "type" discriminator.
• Helper functions is_event, is_request, and is_wire_message provide fast runtime type checking without manual isinstance chains.
• All public symbols are listed in the module's __all__ definition (lines 64–78) and consumed by src/kimi_cli/soul/kimisoul.py and the UI front-ends in src/kimi_cli/ui/*.

Frequently Asked Questions

What is the difference between an Event and a Request in Kimi CLI?

An Event is a unidirectional notification emitted by the runtime to inform the UI about control-flow changes, content streams, or status updates. A Request is sent when the runtime needs explicit input from the client, such as an approval or a tool result, and the runtime blocks until a response is returned.

How does WireMessageEnvelope know which concrete type to deserialize?

The envelope stores a "type" string field that maps directly to the Pydantic model name. When to_wire_message() is called, the envelope uses this discriminator to instantiate the correct class from the WireMessage union. This design lets the JSON-line protocol remain dynamically typed on the wire while preserving static typing inside Python.

Where are Wire message types exported for use across the Kimi CLI codebase?

All canonical types appear in the __all__ list at the end of src/kimi_cli/wire/types.py (lines 64–78). Other packages typically import from kimi_cli.wire, which re-exports the public API, keeping consumer code independent of the internal module layout.

How do you check if a deserialized message is an Event or Request at runtime?

Use the is_event(msg) and is_request(msg) helpers defined at lines 12–25 of src/kimi_cli/wire/types.py. Both functions perform an optimized isinstance check against the Event and Request unions, respectively, and return a boolean that UI dispatch loops can use to route messages correctly.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}