# Events Tools (/aiway/mcp/tools/events)



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 [#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 [#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](/connectors),
then returns the result in the standard `content[]` envelope.

<Mermaid
  chart="`
graph LR
AI[&#x22;AI model / MCP client&#x22;]
MCP[&#x22;MCP connector<br/>POST /mcp&#x22;]
EV{{&#x22;Events channel<br/>(ephemeral)&#x22;}}
ST{{&#x22;Events store channel<br/>(persistent)&#x22;}}
SUB[&#x22;Live subscribers&#x22;]

AI -- &#x22;events_publish&#x22; --> MCP
AI -- &#x22;events_store_publish&#x22; --> MCP
AI -. &#x22;events_store_read / _latest&#x22; .-> MCP
MCP -- publish --> EV
MCP -- &#x22;append (seq)&#x22; --> ST
EV -. deliver .-> SUB
ST -. &#x22;read by seq / time&#x22; .-> MCP

class AI external
class MCP aiway
class EV events
class ST store
class SUB client
`"
/>

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

## events\_publish [#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.

<Tabs groupId="language" items="['curl','C#','Go','Java','Kotlin','Python','Ruby','Rust','Swift','TypeScript']">
  <Tab value="curl">
    ```bash
    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"}
          }
        }
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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)
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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();
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    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()
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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())
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    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
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    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(())
    }
    ```
  </Tab>

  <Tab value="Swift">
    ```swift
    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)")
    ```
  </Tab>

  <Tab value="TypeScript">
    ```typescript
    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();
    ```
  </Tab>
</Tabs>

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

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

## events\_store\_publish [#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.

<Tabs groupId="language" items="['curl','C#','Go','Java','Kotlin','Python','Ruby','Rust','Swift','TypeScript']">
  <Tab value="curl">
    ```bash
    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"}
          }
        }
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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)
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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);
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    val result = client.callTool("events_store_publish", mapOf(
        "channel" to "example-events-store",
        "body" to "Stored event"
    ))

    println("Result: $result")
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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}")
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    result = client.call_tool("events_store_publish", {
      "channel" => "example-events-store",
      "body" => "Stored event",
    })

    puts "Result: #{result}"
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    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:#?}");
    ```
  </Tab>

  <Tab value="Swift">
    ```swift
    let result = try await client.callTool("events_store_publish", arguments: [
        "channel": "example-events-store",
        "body": "Stored event",
    ])

    print("Result: \(result)")
    ```
  </Tab>

  <Tab value="TypeScript">
    ```typescript
    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));
    ```
  </Tab>
</Tabs>

```json
{
  "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 [#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`.

<Tabs groupId="language" items="['curl','C#','Go','Java','Kotlin','Python','Ruby','Rust','Swift','TypeScript']">
  <Tab value="curl">
    ```bash
    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
          }
        }
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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)
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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);
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    val result = client.callTool("events_store_read", mapOf(
        "channel" to "example-events-store",
        "from_sequence" to 1,
        "max_messages" to 10
    ))

    println("Result: $result")
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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}")
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    result = client.call_tool("events_store_read", {
      "channel" => "example-events-store",
      "from_sequence" => 1,
      "max_messages" => 10,
    })

    puts "Result: #{result}"
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    let result = client.call_tool("events_store_read", json!({
        "channel": "example-events-store",
        "from_sequence": 1,
        "max_messages": 10
    })).await?;

    println!("Result: {result:#?}");
    ```
  </Tab>

  <Tab value="Swift">
    ```swift
    let result = try await client.callTool("events_store_read", arguments: [
        "channel": "example-events-store",
        "from_sequence": 1,
        "max_messages": 10,
    ])

    print("Result: \(result)")
    ```
  </Tab>

  <Tab value="TypeScript">
    ```typescript
    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));
    ```
  </Tab>
</Tabs>

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

```json
{
  "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 [#events_store_read_latest]

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

<Tabs groupId="language" items="['curl','C#','Go','Java','Kotlin','Python','Ruby','Rust','Swift','TypeScript']">
  <Tab value="curl">
    ```bash
    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
          }
        }
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    var result = await client.CallToolAsync("events_store_read_latest", new Dictionary<string, object>
    {
        ["channel"] = "example-events-store",
        ["count"] = 3,
    });

    Console.WriteLine($"Result: {result}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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)
    ```
  </Tab>

  <Tab value="Java">
    ```java
    var result = client.callTool(new CallToolRequest(
        "events_store_read_latest",
        Map.of(
            "channel", "example-events-store",
            "count", 3
        )
    ));
    System.out.println(result);
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    val result = client.callTool("events_store_read_latest", mapOf(
        "channel" to "example-events-store",
        "count" to 3
    ))

    println("Result: $result")
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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}")
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    result = client.call_tool("events_store_read_latest", {
      "channel" => "example-events-store",
      "count" => 3,
    })

    puts "Result: #{result}"
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    let result = client.call_tool("events_store_read_latest", json!({
        "channel": "example-events-store",
        "count": 3
    })).await?;

    println!("Result: {result:#?}");
    ```
  </Tab>

  <Tab value="Swift">
    ```swift
    let result = try await client.callTool("events_store_read_latest", arguments: [
        "channel": "example-events-store",
        "count": 3,
    ])

    print("Result: \(result)")
    ```
  </Tab>

  <Tab value="TypeScript">
    ```typescript
    const result = await client.callTool({
      name: "events_store_read_latest",
      arguments: {
        channel: "example-events-store",
        count: 3,
      },
    });
    console.log(JSON.stringify(result, null, 2));
    ```
  </Tab>
</Tabs>

```json
{
  "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 [#parameters]

### events\_publish [#events_publish-1]

| Argument   | Type   | Required | Default | Description                                     |
| ---------- | ------ | -------- | ------- | ----------------------------------------------- |
| `channel`  | string | yes      | —       | Events channel to publish to.                   |
| `body`     | string | yes      | —       | Event payload.                                  |
| `metadata` | string | no       | —       | Optional metadata string attached to the event. |
| `tags`     | object | no       | —       | Optional key/value string tags.                 |

### events\_store\_publish [#events_store_publish-1]

| Argument   | Type   | Required | Default | Description                                     |
| ---------- | ------ | -------- | ------- | ----------------------------------------------- |
| `channel`  | string | yes      | —       | Events store channel to append to.              |
| `body`     | string | yes      | —       | Event payload.                                  |
| `metadata` | string | no       | —       | Optional metadata string attached to the event. |
| `tags`     | object | no       | —       | Optional key/value string tags.                 |

### events\_store\_read [#events_store_read-1]

| Argument        | Type   | Required | Default | Description                               |
| --------------- | ------ | -------- | ------- | ----------------------------------------- |
| `channel`       | string | yes      | —       | Events store channel to read from.        |
| `max_messages`  | number | yes      | —       | Maximum number of events to return.       |
| `from_sequence` | number | no       | —       | Start reading at this sequence number.    |
| `from_time`     | string | no       | —       | Start reading at this RFC 3339 timestamp. |

### events\_store\_read\_latest [#events_store_read_latest-1]

| Argument  | Type   | Required | Default | Description                                         |
| --------- | ------ | -------- | ------- | --------------------------------------------------- |
| `channel` | string | yes      | —       | Events store channel to read from.                  |
| `count`   | number | no       | `10`    | Number of most-recent events to return (max `100`). |

<Callout type="info">
  Channel names beginning with the reserved `_AGENTS_.` prefix are rejected — see
  [Channel resolution](/aiway/mcp/guides/channel-resolution).
</Callout>

## Response [#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](/aiway/mcp/guides/error-handling) for the three failure
layers.

## Related [#related]

<Cards>
  <Card title="Tools overview" href="/aiway/mcp/tools" description="The full 15-tool map and the shared tools/call response shape." />

  <Card title="Queue tools" href="/aiway/mcp/tools/queues" description="queue_send, queue_receive, and queue_peek for durable FIFO queues." />

  <Card title="Tools reference" href="/aiway/mcp/reference/tools-reference" description="Full catalog: arguments, defaults, and response shapes for every tool." />

  <Card title="Error handling" href="/aiway/mcp/guides/error-handling" description="Detect tool errors, JSON-RPC errors, and HTTP failures." />
</Cards>