colmugx/mcp/transport - mooncakes.io

README
U
CallToolResult
U
CompletionResult
U
ContentItem
U
JsonRpcRequest
U
JsonRpcResponse
U
JsonSchema
U
ListPromptsResult
U
ListResourcesResult
U
ListToolsResult
U
MCPClient
U
MCPError
U
MCPHost
U
MCPServer
U
ParamDef
U
Params
U
ReadResourceResult
U
ServerCapabilities
U
ServerInfo
U
Tool
U
ToolCallOutcome
U
ToolDefinition
U
ToolError
U
ToolResult
U
TransportError
V
get_number
V
get_optional_number
V
get_optional_string
V
get_string
V
mcp_server
V
simple_tool
V
tool_fn

colmugx/mcp/transport does not have a README file

Transport

pub(open) trait Transport {
  fn receive(Self) -> String? raise @colmugx/mcp/protocol/types.TransportError
TransportError
  fn send(Self, String) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError
  fn send_notification(Self, @colmugx/mcp/protocol/types.Notification
Notification) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError
  fn send_event(Self, event_type~ : String, data~ : String) -> Unit
  fn supports_streaming(Self) -> Bool
  fn close(Self) -> Unit
}

AnyTransport

source

pub(all) enum AnyTransport {
  Stdio(StdioTransport)
  StdioClient(StdioClientTransport)
  Http(HttpTransport)
  HttpClient(HttpClientTransport)
}

AnyTransport::close

source

fn AnyTransport::close(self : AnyTransport) -> Unit

AnyTransport::receive

source

async fn AnyTransport::receive(self : AnyTransport) -> String? raise @colmugx/mcp/protocol/types.TransportError
TransportError

AnyTransport::send

source

async fn AnyTransport::send(self : AnyTransport, message : String) -> Unit

AnyTransport::send_event

source

fn AnyTransport::send_event(self : AnyTransport, event_type~ : String, data~ : String) -> Unit

AnyTransport::send_notification

source

async fn AnyTransport::send_notification(self : AnyTransport, notification : @colmugx/mcp/protocol/types.Notification
Notification) -> Unit

AnyTransport::supports_streaming

source

fn AnyTransport::supports_streaming(self : AnyTransport) -> Bool

AuthConfig

source

pub struct AuthConfig {
  verify_token : (String) -> Bool
  resource_metadata_url : String
  authorization_servers : Array[String]
  required_scopes : String?
  allowed_origins : Array[String]?
}

Server-side authentication configuration (MCP spec 2026-07-28). Provides Bearer token validation, Protected Resource Metadata, and Origin header validation for DNS rebinding prevention.

AuthConfig::AuthConfig

source

fn AuthConfig::AuthConfig(verify_token~ : (String) -> Bool, resource_metadata_url~ : String, authorization_servers? : Array[String], required_scopes? : String, allowed_origins? : Array[String]) -> AuthConfig

HttpClientTransport

source

pub struct HttpClientTransport {
  base_url : String
  auth_token : String?
  closed : Bool
  pending_responses : @moonbitlang/async/aqueue.Queue
Queue[String]
  tool_header_schemas : Map[String, Array[(Array[String], String)]]
}

HttpClientTransport — MCP client connecting to a remote HTTP server.

Implements the stateless 2026-07-28 Streamable HTTP transport: every JSON-RPC request is its own POST, carrying the required request-metadata headers (MCP-Protocol-Version, Mcp-Method, Mcp-Name). A POST may be answered with either a single JSON object or an SSE stream scoped to that request; both are collected into pending_responses and surfaced one at a time by receive.

The legacy session model (Mcp-Session-Id, GET long-poll SSE, Last-Event-ID resumability, DELETE close) is removed. Server→client notifications now flow on per-request SSE response streams (progress) or subscriptions/listen (change events), not a standalone GET channel.

HttpClientTransport::HttpClientTransport

source

fn HttpClientTransport::HttpClientTransport(base_url : String, auth_token? : String) -> HttpClientTransport

HttpClientTransport::close

source

fn HttpClientTransport::close(self : HttpClientTransport) -> Unit

HttpClientTransport::receive

source

async fn HttpClientTransport::receive(self : HttpClientTransport) -> String? noraise

Drain the next response message (single JSON or one SSE event) queued by send. Blocks asynchronously until a message is available. Returns None when the transport is closed or the underlying queue is closed/empty.

HttpClientTransport::send

source

async fn HttpClientTransport::send(self : HttpClientTransport, message : String) -> Unit

POST a JSON-RPC request to the server. Sets the required request-metadata headers (MCP-Protocol-Version, Mcp-Method, Mcp-Name) and accepts either a single JSON object or an SSE stream as the response. All response messages are pushed onto pending_responses for receive to drain. Cancellation note: every error-translating catch below re-raises when @async.is_being_cancelled() — otherwise a with_timeout cancellation injected at a suspension point would be re-labeled as WriteError/ReadError, losing its identity to the enclosing task group's timeout fallback.

HttpClientTransport::send_event

source

fn HttpClientTransport::send_event(_self : HttpClientTransport, event_type~ : String, data~ : String) -> Unit

Client transport does not push events to server.

HttpClientTransport::send_notification

source

async fn HttpClientTransport::send_notification(self : HttpClientTransport, notification : @colmugx/mcp/protocol/types.Notification
Notification) -> Unit

Send a JSON-RPC notification (no id, fire-and-forget). Per spec §B, a notification POST is answered with 202 Accepted and no body.

HttpClientTransport::set_tool_header_schemas

source

fn HttpClientTransport::set_tool_header_schemas(self : HttpClientTransport, schemas : Map[String, Array[(Array[String], String)]]) -> Unit

Install the x-mcp-header mirror schema set (called by the client layer after validating tools/list tool definitions). Emission itself is done by send when building tools/call request headers.

HttpClientTransport::supports_streaming

source

fn HttpClientTransport::supports_streaming(_self : HttpClientTransport) -> Bool

HttpTransport

source

pub struct HttpTransport {
  port : Int
  endpoint_path : String
  auth : AuthConfig?
  pending_requests : @moonbitlang/async/aqueue.Queue
Queue[(String, @moonbitlang/async/aqueue.Queue
Queue[String])]
  pending_reply_queues : Map[String, @moonbitlang/async/aqueue.Queue
Queue[String]]
}

HttpTransport::HttpTransport

source

fn HttpTransport::HttpTransport(port? : Int, endpoint_path? : String) -> HttpTransport

HttpTransport::close

source

fn HttpTransport::close(self : HttpTransport) -> Unit

HttpTransport::receive

source

async fn HttpTransport::receive(self : HttpTransport) -> String? raise @colmugx/mcp/protocol/types.TransportError
TransportError

HttpTransport::receive_request

source

async fn HttpTransport::receive_request(self : HttpTransport) -> (String, @moonbitlang/async/aqueue.Queue
Queue[String])? raise @colmugx/mcp/protocol/types.TransportError
TransportError

HttpTransport::send

source

async fn HttpTransport::send(self : HttpTransport, message : String) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

HttpTransport::send_event

source

fn HttpTransport::send_event(self : HttpTransport, event_type~ : String, data~ : String) -> Unit

HttpTransport::send_notification

source

fn HttpTransport::send_notification(self : HttpTransport, _notification : @colmugx/mcp/protocol/types.Notification
Notification) -> Unit

Trait-required stub. Server→client notifications over HTTP flow through MCPServer's per-subscription reply handles (subscriptions/listen), not through this transport; the legacy GET-SSE event queue is gone.

HttpTransport::start

source

async fn HttpTransport::start(self : HttpTransport) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

HttpTransport::supports_streaming

source

fn HttpTransport::supports_streaming(_self : HttpTransport) -> Bool

HttpTransport::with_auth

source

fn HttpTransport::with_auth(self : HttpTransport, auth : AuthConfig) -> HttpTransport

JsonRpcKind

source

pub(all) enum JsonRpcKind {
  Notification
  FinalResponse
  Other
} derive(Eq, @moonbitlang/core/debug.Debug
Debug)

Classification of JSON-RPC messages arriving on the server-side reply queue. Used to decide whether a POST response is a single JSON object or an SSE stream.

StdioClientTransport

source

pub struct StdioClientTransport {
  cmd : String
  args : Array[String]
  extra_env : Map[String, String]
  closed : Bool
  reader : @moonbitlang/async/process.ReadFromProcess
ReadFromProcess?
  raw_writer : @moonbitlang/async/process.WriteToProcess
WriteToProcess?
  writer : @moonbitlang/async/io.BufferedWriter
BufferedWriter[@moonbitlang/async/process.WriteToProcess
WriteToProcess]?
}

StdioClientTransport — MCP host/client connecting to a local MCP server via spawned child process.

Follows the MCP specification: the host spawns the server as a subprocess, communicates via newline-delimited JSON-RPC over stdin/stdout pipes. Stderr passes through to the parent for server-side logging.

Two-phase initialization:

new() stores configuration (command, arguments, env)
start(group) creates pipes and spawns the child process

Graceful shutdown (per MCP spec):

Close stdin pipe → signals EOF to child
Child detects EOF and exits
TaskGroup cancel_handler (SIGTERM → 5s → SIGKILL) handles stubborn processes

StdioClientTransport::StdioClientTransport

source

fn StdioClientTransport::StdioClientTransport(cmd~ : String, args? : Array[String], extra_env? : Map[String, String]) -> StdioClientTransport

StdioClientTransport::close

source

fn StdioClientTransport::close(self : StdioClientTransport) -> Unit

Graceful shutdown per MCP spec:

Close stdin writer → signals EOF to child process
Close stdout reader → release pipe resources
Mark transport as closed The TaskGroup's cancel_handler handles forced termination if needed.

StdioClientTransport::receive

source

async fn StdioClientTransport::receive(self : StdioClientTransport) -> String? raise @colmugx/mcp/protocol/types.TransportError
TransportError

Read one JSON-RPC message from the child's stdout. Returns None on EOF (child closed stdout / process exited).

StdioClientTransport::send

source

async fn StdioClientTransport::send(self : StdioClientTransport, message : String) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

Send a JSON-RPC message to the child's stdin. Validates the message, writes it with a newline, and flushes immediately.

StdioClientTransport::send_event

source

fn StdioClientTransport::send_event(_self : StdioClientTransport, event_type~ : String, data~ : String) -> Unit

Client transport does not push events to child process.

StdioClientTransport::send_notification

source

async fn StdioClientTransport::send_notification(self : StdioClientTransport, notification : @colmugx/mcp/protocol/types.Notification
Notification) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

Send a JSON-RPC notification (no id, fire-and-forget) to the child's stdin.

StdioClientTransport::start

source

async fn StdioClientTransport::start(self : StdioClientTransport, group : @moonbitlang/async.TaskGroup
TaskGroup[Unit]) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

Spawn the child process inside the given TaskGroup. Creates stdin/stdout pipes, spawns the process, and wraps the writer in a BufferedWriter (8KB) for efficient I/O.

Must be called before send() / receive().

StdioClientTransport::supports_streaming

source

fn StdioClientTransport::supports_streaming(_self : StdioClientTransport) -> Bool

StdioTransport

source

pub struct StdioTransport {
  closed : Bool
  reader : @moonbitlang/async/stdio.Input
Input
  writer : @moonbitlang/async/io.BufferedWriter
BufferedWriter[@moonbitlang/async/stdio.Output
Output]
}

StdioTransport with buffered I/O for improved performance. Buffering reduces the number of syscalls and context switches, which is critical for high-frequency request/response patterns.

StdioTransport::StdioTransport

source

fn StdioTransport::StdioTransport() -> StdioTransport

StdioTransport::close

source

fn StdioTransport::close(self : StdioTransport) -> Unit

StdioTransport::receive

source

async fn StdioTransport::receive(self : StdioTransport) -> String? raise @colmugx/mcp/protocol/types.TransportError
TransportError

StdioTransport::send

source

async fn StdioTransport::send(self : StdioTransport, message : String) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

StdioTransport::send_event

source

fn StdioTransport::send_event(_self : StdioTransport, event_type~ : String, data~ : String) -> Unit

StdioTransport::send_notification

source

async fn StdioTransport::send_notification(self : StdioTransport, notification : @colmugx/mcp/protocol/types.Notification
Notification) -> Unit raise @colmugx/mcp/protocol/types.TransportError
TransportError

StdioTransport::supports_streaming

source

fn StdioTransport::supports_streaming(_self : StdioTransport) -> Bool

classify_jsonrpc_message

source

fn classify_jsonrpc_message(message : String) -> JsonRpcKind

Classify a serialized JSON-RPC message for response-mode selection.

decode_base64

source

fn decode_base64(s : String) -> Bytes?

Decode a standard Base64 string back to raw bytes. Returns None on invalid input (fail-closed).

decode_base64_sentinel

source

fn decode_base64_sentinel(value : String) -> String?

Decode a Base64 sentinel value. Returns None if the value is not in sentinel form or if decoding fails (fail-closed).

encode_base64

source

fn encode_base64(bytes : Bytes) -> String

Encode raw bytes to a standard Base64 string (with padding).

encode_base64_sentinel

source

fn encode_base64_sentinel(value : String) -> String

Encode a string as a Base64 sentinel value: =?base64?<b64>?=. The input is first encoded as UTF-8 bytes.

encode_header_value

source

fn encode_header_value(value : String) -> String

Encode a header value according to the spec: plain when safe, otherwise Base64-sentinel encoded.

extract_argument_value

source

fn extract_argument_value(body : String, path : Array[String]) -> String?

Extract the value at a property path inside params.arguments for tools/call requests. Returns None when the path is absent or the final value is null. Converts string, safe integer, and boolean values to their header string representations.

is_header_safe

source

fn is_header_safe(value : String) -> Bool

Returns true if value can be transmitted as a plain HTTP header value: every character is visible ASCII (0x21–0x7E), space (0x20), or tab (0x09); it has no leading/trailing whitespace; and it does not match the sentinel pattern (to avoid ambiguity).

is_notification

source

fn is_notification(body : Json) -> Bool

Returns true when the JSON body is a JSON-RPC notification (method present, id absent).

is_valid_header_characters

source

fn is_valid_header_characters(value : String) -> Bool

Returns true if value contains only characters permitted in a non- sentinel HTTP header value: visible ASCII (0x21–0x7E), space (0x20), or tab (0x09). CR and LF are implicitly rejected because they are not in the allowed range.

jsonrpc_error_code

source

fn jsonrpc_error_code(message : String) -> Int?

Extract the JSON-RPC error code from a serialized response, if any.

looks_like_sentinel

source

fn looks_like_sentinel(value : String) -> Bool

Returns true if the value matches the Base64 sentinel pattern.

read_sse_event

source

async fn read_sse_event(client : @moonbitlang/async/http.Client
Client) -> (String?, String?) raise @colmugx/mcp/protocol/types.TransportError
TransportError

SSE (Server-Sent Events) scanner for streamable HTTP transports.

read_sse_event consumes exactly one event — or detects EOF — from a streaming HTTP response body. Tokenization runs through lexscan over an @lexbuf.AsyncLexbuf, so the per-line matching stays on the compiler-optimized regex path and patterns may span refill-chunk boundaries.

Semantics follow the WHATWG SSE spec: lines end with \n, \r\n, or \r; a blank line dispatches the accumulated event; data:/id: values strip a single leading space; comments and unknown fields are ignored; consecutive data: lines join with \n; an unterminated final line at EOF is dispatched as if the terminator were present.

sse_event_line

source

fn sse_event_line(json : String) -> String

Format a JSON-RPC payload as one SSE event line.

unsupported_protocol_version_error

source

fn unsupported_protocol_version_error(requested : String) -> String

Build an UnsupportedProtocolVersion (-32022) JSON-RPC error response body. The response has no id because the failure is transport-level.

validate_jsonrpc_message

source

fn validate_jsonrpc_message(message : String) -> Result[Unit, @colmugx/mcp/protocol/types.MCPError
MCPError]

Source Files

Site source Report issue Packages Build queue Skills Statistics

#Transport

#AnyTransport

#AnyTransport::close

#AnyTransport::receive

#AnyTransport::send

#AnyTransport::send_event

#AnyTransport::send_notification

#AnyTransport::supports_streaming

#AuthConfig

#AuthConfig::AuthConfig

#HttpClientTransport

#HttpClientTransport::HttpClientTransport

#HttpClientTransport::close

#HttpClientTransport::receive

#HttpClientTransport::send

#HttpClientTransport::send_event

#HttpClientTransport::send_notification

#HttpClientTransport::set_tool_header_schemas

#HttpClientTransport::supports_streaming

#HttpTransport

#HttpTransport::HttpTransport

#HttpTransport::close

#HttpTransport::receive

#HttpTransport::receive_request

#HttpTransport::send

#HttpTransport::send_event

#HttpTransport::send_notification

#HttpTransport::start

#HttpTransport::supports_streaming

#HttpTransport::with_auth

#JsonRpcKind

#StdioClientTransport

#StdioClientTransport::StdioClientTransport

#StdioClientTransport::close

#StdioClientTransport::receive

#StdioClientTransport::send

#StdioClientTransport::send_event

#StdioClientTransport::send_notification

#StdioClientTransport::start

#StdioClientTransport::supports_streaming

#StdioTransport

#StdioTransport::StdioTransport

#StdioTransport::close

#StdioTransport::receive

#StdioTransport::send

#StdioTransport::send_event

#StdioTransport::send_notification

#StdioTransport::supports_streaming

#classify_jsonrpc_message

#decode_base64

#decode_base64_sentinel

#encode_base64

#encode_base64_sentinel

#encode_header_value

#extract_argument_value

#is_header_safe

#is_notification

#is_valid_header_characters

#jsonrpc_error_code

#looks_like_sentinel

#read_sse_event

#sse_event_line

#unsupported_protocol_version_error

#validate_jsonrpc_message

Source Files

Transport

AnyTransport

AnyTransport::close

AnyTransport::receive

AnyTransport::send

AnyTransport::send_event

AnyTransport::send_notification

AnyTransport::supports_streaming

AuthConfig

AuthConfig::AuthConfig

HttpClientTransport

HttpClientTransport::HttpClientTransport

HttpClientTransport::close

HttpClientTransport::receive

HttpClientTransport::send

HttpClientTransport::send_event

HttpClientTransport::send_notification

HttpClientTransport::set_tool_header_schemas

HttpClientTransport::supports_streaming

HttpTransport

HttpTransport::HttpTransport

HttpTransport::close

HttpTransport::receive

HttpTransport::receive_request

HttpTransport::send

HttpTransport::send_event

HttpTransport::send_notification

HttpTransport::start

HttpTransport::supports_streaming

HttpTransport::with_auth

JsonRpcKind

StdioClientTransport

StdioClientTransport::StdioClientTransport

StdioClientTransport::close

StdioClientTransport::receive

StdioClientTransport::send

StdioClientTransport::send_event

StdioClientTransport::send_notification

StdioClientTransport::start

StdioClientTransport::supports_streaming

StdioTransport

StdioTransport::StdioTransport

StdioTransport::close

StdioTransport::receive

StdioTransport::send

StdioTransport::send_event

StdioTransport::send_notification

StdioTransport::supports_streaming

classify_jsonrpc_message

decode_base64

decode_base64_sentinel

encode_base64

encode_base64_sentinel

encode_header_value

extract_argument_value

is_header_safe

is_notification

is_valid_header_characters

jsonrpc_error_code

looks_like_sentinel

read_sse_event

sse_event_line

unsupported_protocol_version_error

validate_jsonrpc_message