The channel-management tools let an AI model discover what messaging surfaces exist on a KubeMQ server before it sends, publishes, or queries anything. channel_list enumerates channels (optionally filtered), and channel_info returns live metadata for one channel.

Overview

Most MCP tools act on a channel you already know — queue_send needs a queue name, events_publish needs an events channel. The two read-only channel-management tools close that gap: they let the model explore the broker's namespace and confirm a channel's type and activity before acting on it.

Tool	Purpose	Required arguments
`channel_list`	List channels, optionally filtered by type or name pattern	(none)
`channel_info`	Return metadata for one specific channel	`channel`, `type`

A channel type is one of queues, events, events_store, commands, or queries — the five KubeMQ messaging patterns. Both tools accept the type to scope the lookup. Reserved channels (those under the _AGENTS_. prefix used by the agent bridge) are internal and are not addressable through these tools.

Both tools are read-only discovery operations — they never create, delete, or modify a channel. Channels in KubeMQ are created implicitly on first use, so channel_list reflects channels that already carry traffic.

channel_list

List all channels on the server, or narrow the result by channel type, by a name pattern, or both. With no arguments it returns every known channel.

Input schema

{
  "type": "object",
  "required": [],
  "properties": {
    "type": {
      "type": "string",
      "description": "Filter by channel type (queues, events, events_store, commands, queries)."
    },
    "pattern": {
      "type": "string",
      "description": "Filter by channel name pattern or prefix."
    }
  }
}

Argument	Type	Required	Default	Description
`type`	string	no	—	Filter by channel type: `queues`, `events`, `events_store`, `commands`, `queries`
`pattern`	string	no	—	Filter by channel name pattern or prefix

Output

The tool result wraps a JSON array of channel summaries in the standard MCP content[]/isError envelope. Each element carries the channel name, type, and is_active flag.

{
  "jsonrpc": "2.0",
  "id": 11,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"name\":\"example-queue\",\"type\":\"queues\",\"is_active\":true},{\"name\":\"example-events\",\"type\":\"events\",\"is_active\":true}]"
      }
    ],
    "isError": false
  }
}

An empty list ([]) is a normal successful result, not an error.

Usage

# List all channels (no filter)
curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json, text/event-stream' \
  -H 'MCP-Protocol-Version: 2025-11-25' \
  -d '{
    "jsonrpc": "2.0",
    "id": 11,
    "method": "tools/call",
    "params": {
      "name": "channel_list",
      "arguments": {}
    }
  }'

# Filter by type
# "arguments": { "type": "queues" }
# Filter by name pattern
# "arguments": { "pattern": "example-" }

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:      "channel_list",
        Arguments: map[string]any{},
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)

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

        # List all channels (no filter)
        result = await session.call_tool("channel_list", {})

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

        # To filter by type, use: {"type": "queues"}
        # To filter by pattern, use: {"pattern": "example-"}

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);

// List all channels (no filter)
const result = await client.callTool({
  name: "channel_list",
  arguments: {},
});
console.log(JSON.stringify(result, null, 2));

// To filter by type: { type: "queues" }
// To filter by pattern: { pattern: "example-" }

await client.close();

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();

// List all channels (no filter)
var result = client.callTool(new CallToolRequest(
    "channel_list",
    Map.of()
));
System.out.println(result);

// To filter by type: Map.of("type", "queues")
// To filter by pattern: Map.of("pattern", "example-")

client.closeGracefully();

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("channel_list", new Dictionary<string, object>());

Console.WriteLine($"Result: {result}");

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("channel_list", emptyMap())

println("Result: $result")

client.close()
httpClient.close()

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("channel_list", {})

puts "Result: #{result}"

client.close

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("channel_list", json!({})).await?;

println!("Result: {result:#?}");

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("channel_list", arguments: [:])

print("Result: \(result)")

channel_info

Return metadata for a single channel. Both channel (the name) and type are required — the type scopes the lookup to the right messaging pattern.

Input schema

{
  "type": "object",
  "required": ["channel", "type"],
  "properties": {
    "channel": {
      "type": "string",
      "description": "Channel name to get information for."
    },
    "type": {
      "type": "string",
      "description": "Channel type (queues, events, events_store, commands, queries)."
    }
  }
}

Argument	Type	Required	Default	Description
`channel`	string	yes	—	Channel name to inspect
`type`	string	yes	—	Channel type: `queues`, `events`, `events_store`, `commands`, `queries`

Output

The result wraps a single JSON object describing the channel. Alongside name, type, and is_active, queue-style channels report live incoming/outgoing message counts.

{
  "jsonrpc": "2.0",
  "id": 12,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"name\":\"example-queue\",\"type\":\"queues\",\"is_active\":true,\"incoming\":5,\"outgoing\":3}"
      }
    ],
    "isError": false
  }
}

Requesting a channel that does not exist returns a tool error (isError: true) rather than a transport-level failure.

Usage

curl -X POST http://localhost:9090/mcp \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json, text/event-stream' \
  -H 'MCP-Protocol-Version: 2025-11-25' \
  -d '{
    "jsonrpc": "2.0",
    "id": 12,
    "method": "tools/call",
    "params": {
      "name": "channel_info",
      "arguments": {
        "channel": "example-queue",
        "type": "queues"
      }
    }
  }'

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: "channel_info",
        Arguments: map[string]any{
            "channel": "example-queue",
            "type":    "queues",
        },
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Result: %+v\n", result)

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("channel_info", {
            "channel": "example-queue",
            "type": "queues",
        })

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

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: "channel_info",
  arguments: {
    channel: "example-queue",
    type: "queues",
  },
});
console.log(JSON.stringify(result, null, 2));

await client.close();

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(
    "channel_info",
    Map.of(
        "channel", "example-queue",
        "type", "queues"
    )
));
System.out.println(result);

client.closeGracefully();

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("channel_info", new Dictionary<string, object>
{
    ["channel"] = "example-queue",
    ["type"] = "queues",
});

Console.WriteLine($"Result: {result}");

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("channel_info", mapOf(
    "channel" to "example-queue",
    "type" to "queues"
))

println("Result: $result")

client.close()
httpClient.close()

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("channel_info", {
  "channel" => "example-queue",
  "type" => "queues",
})

puts "Result: #{result}"

client.close

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("channel_info", json!({
    "channel": "example-queue",
    "type": "queues"
})).await?;

println!("Result: {result:#?}");

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("channel_info", arguments: [
    "channel": "example-queue",
    "type": "queues",
])

print("Result: \(result)")

Error handling

Condition	Result
Invalid or malformed arguments	JSON-RPC `-32602` Invalid Params
`channel_info` on a non-existent channel	Tool error — `isError: true` in the result
`channel_list` with no matches	Successful result with an empty array `[]`

channel_list has no tool-specific failures: an empty list is a normal, successful response. For the full JSON-RPC error catalog, see Error codes.

Channel Management Tools

Overview

channel_list

Input schema

Output

Usage

channel_info

Input schema

Output

Usage

Error handling

Tools overview

Queue tools

Tools reference

On this page