KubeMQ
AiwayMCPTools

Events Tools

Publish ephemeral and persistent events and read the events store as MCP tools — events_publish, events_store_publish, and events_store_read on KubeMQ.

The events tools let an AI model publish events to KubeMQ and read them back from the events store, all over the Model Context Protocol. They cover both the fire-and-forget pub/sub pattern and the persistent, replayable events store.

Overview

The MCP connector exposes four events tools, split across two delivery models:

  • events_publish — fire-and-forget pub/sub. The event is delivered to whatever subscribers are live at that instant and is not stored; if no one is listening, it is lost.
  • events_store_publish — persistent publish. The event is appended to the events store with a sequence number and can be re-read later.
  • events_store_read — read stored events starting from a sequence number or a timestamp.
  • events_store_read_latest — return the most recent N stored events.

Use events_publish for live notifications where missed messages are acceptable, and the events-store tools when a model needs durable history it can replay — for example, reading recent events to build context before acting.

How it works

Every tool is a tools/call JSON-RPC request. The connector translates the call into a native KubeMQ events or events-store operation over the Array, then returns the result in the standard content[] envelope.

Ephemeral events fan out to live subscribers; events-store events are appended with a sequence number and read back on demand.

events_publish

Publish a fire-and-forget event to an events channel. The call returns as soon as the event is accepted — there is no stored copy and no per-subscriber acknowledgement.

curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 5,
    "method": "tools/call",
    "params": {
      "name": "events_publish",
      "arguments": {
        "channel": "example-events",
        "body": "Event data",
        "metadata": "event-meta",
        "tags": {"source": "mcp-example"}
      }
    }
  }'
using ModelContextProtocol.Client;

var url = Environment.GetEnvironmentVariable("KUBEMQ_MCP_URL") ?? "http://localhost:9090";
var transport = new HttpClientTransport(new HttpClientTransportOptions { Endpoint = new Uri($"{url}/mcp") });
await using var client = await McpClientFactory.CreateAsync(transport);

var result = await client.CallToolAsync("events_publish", new Dictionary<string, object>
{
    ["channel"] = "example-events",
    ["body"] = "Event data",
    ["metadata"] = "event-meta",
    ["tags"] = new Dictionary<string, string> { ["source"] = "mcp-example" },
});

Console.WriteLine($"Result: {result}");
url := os.Getenv("KUBEMQ_MCP_URL")
if url == "" {
    url = "http://localhost:9090"
}

c, err := client.NewStreamableHttpClient(url + "/mcp")
if err != nil {
    log.Fatal(err)
}
defer c.Close()

ctx := context.Background()
if err := c.Start(ctx); err != nil {
    log.Fatal(err)
}

result, err := c.CallTool(ctx, mcp.CallToolRequest{
    Params: mcp.CallToolParams{
        Name: "events_publish",
        Arguments: map[string]any{
            "channel":  "example-events",
            "body":     "Event data",
            "metadata": "event-meta",
            "tags":     map[string]any{"source": "mcp-example"},
        },
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)
String url = System.getenv().getOrDefault("KUBEMQ_MCP_URL", "http://localhost:9090");
var transport = HttpClientStreamableHttpTransport.builder(url).endpoint("/mcp").build();
var client = McpClient.sync(transport).build();
client.initialize();

var result = client.callTool(new CallToolRequest(
    "events_publish",
    Map.of(
        "channel", "example-events",
        "body", "Event data",
        "metadata", "event-meta",
        "tags", Map.of("source", "mcp-example")
    )
));
System.out.println(result);

client.closeGracefully();
val url = System.getenv("KUBEMQ_MCP_URL") ?: "http://localhost:9090"

val httpClient = HttpClient { install(SSE) }
val transport = StreamableHttpClientTransport(client = httpClient, url = "$url/mcp")
val client = Client(clientInfo = Implementation(name = "kubemq-mcp-kotlin-example", version = "1.0.0"))
client.connect(transport)

val result = client.callTool("events_publish", mapOf(
    "channel" to "example-events",
    "body" to "Event data"
))

println("Result: $result")

client.close()
httpClient.close()
import asyncio
import os

from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession

KUBEMQ_MCP_URL = os.environ.get("KUBEMQ_MCP_URL", "http://localhost:9090")


async def main():
    async with streamablehttp_client(f"{KUBEMQ_MCP_URL}/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            result = await session.call_tool("events_publish", {
                "channel": "example-events",
                "body": "Event data",
                "metadata": "event-meta",
                "tags": {"source": "mcp-example"},
            })

            print(f"IsError: {result.isError}")
            for content in result.content:
                print(f"Result: {content.text}")


if __name__ == "__main__":
    asyncio.run(main())
require "mcp"

url = ENV.fetch("KUBEMQ_MCP_URL", "http://localhost:9090")

client = MCP::Client.new(
  transport: MCP::Transport::StreamableHTTP.new("#{url}/mcp"),
  name: "kubemq-mcp-ruby-example",
  version: "1.0.0"
)
client.initialize_handshake

result = client.call_tool("events_publish", {
  "channel" => "example-events",
  "body" => "Event data",
})

puts "Result: #{result}"

client.close
use rmcp::transport::streamable_http::StreamableHttpClientTransport;
use rmcp::service::RunService;
use serde_json::json;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let url = std::env::var("KUBEMQ_MCP_URL")
        .unwrap_or_else(|_| "http://localhost:9090".to_string());

    let transport = StreamableHttpClientTransport::from_uri(format!("{url}/mcp"))?;
    let client = ().serve(transport).await?;

    let result = client.call_tool("events_publish", json!({
        "channel": "example-events",
        "body": "Event data",
        "metadata": "event-meta",
        "tags": {"source": "mcp-example"}
    })).await?;

    println!("Result: {result:#?}");
    Ok(())
}
import Foundation
import MCP

let url = ProcessInfo.processInfo.environment["KUBEMQ_MCP_URL"] ?? "http://localhost:9090"

let transport = HTTPClientTransport(endpoint: URL(string: "\(url)/mcp")!, streaming: true)
let client = Client(name: "kubemq-mcp-swift-example", version: "1.0.0")
try await client.connect(transport: transport)

let result = try await client.callTool("events_publish", arguments: [
    "channel": "example-events",
    "body": "Event data",
])

print("Result: \(result)")
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const KUBEMQ_MCP_URL = process.env.KUBEMQ_MCP_URL || "http://localhost:9090";

const transport = new StreamableHTTPClientTransport(
  new URL(`${KUBEMQ_MCP_URL}/mcp`)
);
const client = new Client({ name: "kubemq-mcp-ts-example", version: "1.0.0" });
await client.connect(transport);

const result = await client.callTool({
  name: "events_publish",
  arguments: {
    channel: "example-events",
    body: "Event data",
    metadata: "event-meta",
    tags: { source: "mcp-example" },
  },
});
console.log(JSON.stringify(result, null, 2));

await client.close();

A successful publish returns a confirmation message in the standard envelope:

{
  "jsonrpc": "2.0",
  "id": 5,
  "result": {
    "content": [{ "type": "text", "text": "Event published successfully to channel 'example-events'" }],
    "isError": false
  }
}

events_store_publish

Publish a persistent event. The event is appended to the events store, assigned a sequence number, and remains available for re-reading by later events_store_read and events_store_read_latest calls.

curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 6,
    "method": "tools/call",
    "params": {
      "name": "events_store_publish",
      "arguments": {
        "channel": "example-events-store",
        "body": "Stored event data",
        "metadata": "store-meta",
        "tags": {"source": "mcp-example"}
      }
    }
  }'
var result = await client.CallToolAsync("events_store_publish", new Dictionary<string, object>
{
    ["channel"] = "example-events-store",
    ["body"] = "Stored event data",
    ["metadata"] = "store-meta",
    ["tags"] = new Dictionary<string, string> { ["source"] = "mcp-example" },
});

Console.WriteLine($"Result: {result}");
result, err := c.CallTool(ctx, mcp.CallToolRequest{
    Params: mcp.CallToolParams{
        Name: "events_store_publish",
        Arguments: map[string]any{
            "channel":  "example-events-store",
            "body":     "Stored event data",
            "metadata": "store-meta",
            "tags":     map[string]any{"source": "mcp-example"},
        },
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)
var result = client.callTool(new CallToolRequest(
    "events_store_publish",
    Map.of(
        "channel", "example-events-store",
        "body", "Stored event data",
        "metadata", "store-meta",
        "tags", Map.of("source", "mcp-example")
    )
));
System.out.println(result);
val result = client.callTool("events_store_publish", mapOf(
    "channel" to "example-events-store",
    "body" to "Stored event"
))

println("Result: $result")
result = await session.call_tool("events_store_publish", {
    "channel": "example-events-store",
    "body": "Stored event data",
    "metadata": "store-meta",
    "tags": {"source": "mcp-example"},
})

print(f"IsError: {result.isError}")
for content in result.content:
    print(f"Result: {content.text}")
result = client.call_tool("events_store_publish", {
  "channel" => "example-events-store",
  "body" => "Stored event",
})

puts "Result: #{result}"
let result = client.call_tool("events_store_publish", json!({
    "channel": "example-events-store",
    "body": "Stored event data",
    "metadata": "store-meta",
    "tags": {"source": "mcp-example"}
})).await?;

println!("Result: {result:#?}");
let result = try await client.callTool("events_store_publish", arguments: [
    "channel": "example-events-store",
    "body": "Stored event",
])

print("Result: \(result)")
const result = await client.callTool({
  name: "events_store_publish",
  arguments: {
    channel: "example-events-store",
    body: "Stored event data",
    metadata: "store-meta",
    tags: { source: "mcp-example" },
  },
});
console.log(JSON.stringify(result, null, 2));
{
  "jsonrpc": "2.0",
  "id": 6,
  "result": {
    "content": [{ "type": "text", "text": "Event published successfully to events store channel 'example-events-store'" }],
    "isError": false
  }
}

events_store_read

Read stored events starting from a position. Provide from_sequence to start at a sequence number, or from_time to start at an RFC 3339 timestamp, and cap the result with max_messages.

curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 7,
    "method": "tools/call",
    "params": {
      "name": "events_store_read",
      "arguments": {
        "channel": "example-events-store",
        "from_sequence": 1,
        "max_messages": 10
      }
    }
  }'
var result = await client.CallToolAsync("events_store_read", new Dictionary<string, object>
{
    ["channel"] = "example-events-store",
    ["from_sequence"] = 1,
    ["max_messages"] = 10,
});

Console.WriteLine($"Result: {result}");
result, err := c.CallTool(ctx, mcp.CallToolRequest{
    Params: mcp.CallToolParams{
        Name: "events_store_read",
        Arguments: map[string]any{
            "channel":       "example-events-store",
            "from_sequence": 1,
            "max_messages":  10,
        },
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)
var result = client.callTool(new CallToolRequest(
    "events_store_read",
    Map.of(
        "channel", "example-events-store",
        "from_sequence", 1,
        "max_messages", 10
    )
));
System.out.println(result);
val result = client.callTool("events_store_read", mapOf(
    "channel" to "example-events-store",
    "from_sequence" to 1,
    "max_messages" to 10
))

println("Result: $result")
result = await session.call_tool("events_store_read", {
    "channel": "example-events-store",
    "from_sequence": 1,
    "max_messages": 10,
})

print(f"IsError: {result.isError}")
for content in result.content:
    print(f"Result: {content.text}")
result = client.call_tool("events_store_read", {
  "channel" => "example-events-store",
  "from_sequence" => 1,
  "max_messages" => 10,
})

puts "Result: #{result}"
let result = client.call_tool("events_store_read", json!({
    "channel": "example-events-store",
    "from_sequence": 1,
    "max_messages": 10
})).await?;

println!("Result: {result:#?}");
let result = try await client.callTool("events_store_read", arguments: [
    "channel": "example-events-store",
    "from_sequence": 1,
    "max_messages": 10,
])

print("Result: \(result)")
const result = await client.callTool({
  name: "events_store_read",
  arguments: {
    channel: "example-events-store",
    from_sequence: 1,
    max_messages: 10,
  },
});
console.log(JSON.stringify(result, null, 2));

The result text is a JSON array of stored events, each carrying its body, metadata, sequence, and timestamp:

{
  "jsonrpc": "2.0",
  "id": 7,
  "result": {
    "content": [{ "type": "text", "text": "[{\"body\":\"Stored event data\",\"metadata\":\"store-meta\",\"sequence\":1,\"timestamp\":\"2026-04-06T12:00:00Z\"}]" }],
    "isError": false
  }
}

events_store_read_latest

Return the most recent events from the store. Set count to choose how many to read back, newest first.

curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -d '{
    "jsonrpc": "2.0",
    "id": 8,
    "method": "tools/call",
    "params": {
      "name": "events_store_read_latest",
      "arguments": {
        "channel": "example-events-store",
        "count": 3
      }
    }
  }'
var result = await client.CallToolAsync("events_store_read_latest", new Dictionary<string, object>
{
    ["channel"] = "example-events-store",
    ["count"] = 3,
});

Console.WriteLine($"Result: {result}");
result, err := c.CallTool(ctx, mcp.CallToolRequest{
    Params: mcp.CallToolParams{
        Name: "events_store_read_latest",
        Arguments: map[string]any{
            "channel": "example-events-store",
            "count":   3,
        },
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)
var result = client.callTool(new CallToolRequest(
    "events_store_read_latest",
    Map.of(
        "channel", "example-events-store",
        "count", 3
    )
));
System.out.println(result);
val result = client.callTool("events_store_read_latest", mapOf(
    "channel" to "example-events-store",
    "count" to 3
))

println("Result: $result")
result = await session.call_tool("events_store_read_latest", {
    "channel": "example-events-store",
    "count": 3,
})

print(f"IsError: {result.isError}")
for content in result.content:
    print(f"Result: {content.text}")
result = client.call_tool("events_store_read_latest", {
  "channel" => "example-events-store",
  "count" => 3,
})

puts "Result: #{result}"
let result = client.call_tool("events_store_read_latest", json!({
    "channel": "example-events-store",
    "count": 3
})).await?;

println!("Result: {result:#?}");
let result = try await client.callTool("events_store_read_latest", arguments: [
    "channel": "example-events-store",
    "count": 3,
])

print("Result: \(result)")
const result = await client.callTool({
  name: "events_store_read_latest",
  arguments: {
    channel: "example-events-store",
    count: 3,
  },
});
console.log(JSON.stringify(result, null, 2));
{
  "jsonrpc": "2.0",
  "id": 8,
  "result": {
    "content": [{ "type": "text", "text": "[{\"body\":\"Stored event 3\",\"sequence\":3},{\"body\":\"Stored event 2\",\"sequence\":2},{\"body\":\"Stored event 1\",\"sequence\":1}]" }],
    "isError": false
  }
}

Parameters

events_publish

ArgumentTypeRequiredDefaultDescription
channelstringyes—Events channel to publish to.
bodystringyes—Event payload.
metadatastringno—Optional metadata string attached to the event.
tagsobjectno—Optional key/value string tags.

events_store_publish

ArgumentTypeRequiredDefaultDescription
channelstringyes—Events store channel to append to.
bodystringyes—Event payload.
metadatastringno—Optional metadata string attached to the event.
tagsobjectno—Optional key/value string tags.

events_store_read

ArgumentTypeRequiredDefaultDescription
channelstringyes—Events store channel to read from.
max_messagesnumberyes—Maximum number of events to return.
from_sequencenumberno—Start reading at this sequence number.
from_timestringno—Start reading at this RFC 3339 timestamp.

events_store_read_latest

ArgumentTypeRequiredDefaultDescription
channelstringyes—Events store channel to read from.
countnumberno10Number of most-recent events to return (max 100).

Channel names beginning with the reserved _AGENTS_. prefix are rejected — see Channel resolution.

Response

Publish tools return a single text confirmation in the content[] envelope. Read tools return a text block whose text is a JSON array of stored events. A failed call sets isError: true and carries the message in the same block — see Error handling for the three failure layers.

Tools overview

The full 15-tool map and the shared tools/call response shape.

Queue tools

queue_send, queue_receive, and queue_peek for durable FIFO queues.

Tools reference

Full catalog: arguments, defaults, and response shapes for every tool.

Error handling

Detect tool errors, JSON-RPC errors, and HTTP failures.

Was this page helpful?

Queue Tools

MCP tools for durable point-to-point queue messaging — queue_send, queue_receive, and queue_peek over the tools/call method.

Command & Query Tools

Use the command_send and query_send MCP tools to make synchronous request/reply calls into KubeMQ from an AI model.

On this page

OverviewHow it worksevents_publishevents_store_publishevents_store_readevents_store_read_latestParametersevents_publishevents_store_publishevents_store_readevents_store_read_latestResponseRelated