Core event bus quickstart#

This page runs Net's low-level Rust event bus in one process, then shows the transport boundary for a real mesh. If you are building an application that needs capability discovery and invocation, start with the Rust SDK quickstart instead.

Every snippet below is compiled in CI as examples/docs_quickstart.rs.

The examples are in Rust because this page documents the Rust-native core crate. Use the language-specific SDK pages for Node.js, Python, and Go; their lifecycles and coverage are not identical.

First, which crate?#

Net ships two Rust entry points. They are layers rather than alternatives:

Start with the SDK for capability discovery, typed invocation, daemons, and tool calling. Use this page when embedding the bus, writing an adapter, or tuning the ingest path.

The two share one runtime: net_sdk::Net is an EventBus with a friendlier surface, and both crates pin to the same version.

Install#

cargo add net-mesh tokio --features tokio/macros,tokio/rt-multi-thread

The crate name on crates.io is net-mesh; you import it as net. Default features compile the full stack — mesh transport, NAT traversal, CortEX, MeshDB, MeshOS, Dataforts. You can pare that down later if you want a smaller build.

Publish#

use std::sync::atomic::Ordering;
use net::{Event, EventBus, EventBusConfig};
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let bus = EventBus::new(EventBusConfig::default()).await?;
 
    bus.ingest(Event::from_str(r#"{"token": "hello", "index": 0}"#)?)?;
    bus.ingest(Event::from_str(r#"{"token": "world", "index": 1}"#)?)?;
 
    // Ingest is non-blocking; flush waits for the batch workers to drain.
    bus.flush().await?;
 
    let stats = bus.stats();
    println!(
        "ingested={} dispatched={} dropped={}",
        stats.events_ingested.load(Ordering::Relaxed),
        stats.events_dispatched.load(Ordering::Relaxed),
        stats.events_dropped.load(Ordering::Relaxed),
    );
 
    bus.shutdown().await?;
    Ok(())
}

Run it and you'll see ingested=2 dispatched=2 dropped=0. That's the whole ingest path: construct, ingest, flush, shutdown.

A few things worth knowing about what just happened:

• EventBusConfig::default() gives you a single-node bus backed by the no-op adapter, which accepts batches and discards them. It's the right shape for benchmarking the ingest path and the wrong shape for anything else.
• bus.ingest() is non-blocking. It hashes the event onto a shard and returns; a background worker drains the shard into the adapter. Throughput depends on the adapter, payload, shard configuration, durability policy, build, and hardware.
• Because ingest is asynchronous, a receipt means accepted into the local ring buffer — not delivered, and not processed. Under backpressure it can drop, which is why events_dropped is worth printing. Submitted Is Not Completed is the long version of that distinction.
• bus.shutdown() drains in-flight ingests, flushes everything to the adapter, and stops the workers cleanly. Calling it is the contract — dropping the bus without shutting down will lose anything still in the ring buffer.

Reading events back#

bus.poll() is the cursor-based consumer, and it reads through the adapter rather than out of the local ring buffer:

use net::ConsumeRequest;
 
let response = bus.poll(ConsumeRequest::new(100)).await?;
for event in response.events {
    // `raw` is the payload as `Bytes` — the bus never assumes it's text.
    println!("{}", String::from_utf8_lossy(&event.raw));
}

Run that against the configuration above and it returns zero events — the no-op adapter had nothing to hand back. That is not a bug you can configure away in one process: the core crate's readable adapters are Redis, JetStream, and the mesh itself, so reading your own events back means choosing one of them. The mesh section below is the usual answer; --features redis or --features jetstream are the others.

Pass from(...) on the request to resume from a cursor, and filter(...) to narrow what comes back.

Add a filter#

Most consumers don't want every event on the bus. Filters are JSON predicates evaluated against the event payload:

use net::Filter;
use serde_json::json;
 
let request = ConsumeRequest::new(100)
    .filter(Filter::eq("token", json!("hello")));
 
let response = bus.poll(request).await?;

Filter values are serde_json::Value, so json! is the shortest way to write one. For several conditions at once, FilterBuilder accumulates them and build_and() / build_or() combine them.

The filter DSL covers existence, equality, numeric comparisons, string matching, and semver — the full grammar lives in the filter reference.

Switch to the mesh#

Everything above runs in one process. To turn the same code into a real distributed bus, you swap the adapter:

use net::adapter::net::{NetAdapterConfig, StaticKeypair};
use net::{AdapterConfig, EventBus, EventBusConfig};
 
// Both sides share a pre-shared key; the responder owns a static keypair
// and the initiator must already know its public half.
let psk = [0x42u8; 32];
let responder = StaticKeypair::generate();
 
let adapter = NetAdapterConfig::initiator(
    "0.0.0.0:7777".parse()?,   // bind
    "10.0.0.2:7777".parse()?,  // peer
    psk,
    responder.public,
);
 
let config = EventBusConfig::builder()
    .adapter(AdapterConfig::Net(Box::new(adapter)))
    .build()?;
 
let bus = EventBus::new(config).await?;

Once configured, ingestion and consumption work identically — ingest() publishes onto the mesh, poll() receives from it.

Two things this snippet makes explicit that are easy to skip past. A Net-adapter link is a Noise session between two named peers, so it is not symmetric: one side is the initiator and must already hold the other's static public key, the other is the responder and holds the keypair. And the key material is yours to distribute — the transport encrypts and authenticates every frame for you, but nothing here invents a PSK or discovers a peer on your behalf.

If what you actually want is a node that finds its peers by capability rather than by address, that's the Mesh surface in the SDK, not the raw adapter — see the Rust SDK quickstart.

That's the part that takes longer than five minutes to fully explore — channel naming, visibility scopes, durable persistence, capability-based authorization — but the call shape never changes. Once you have the loop above working locally, the rest is configuration.

The agentic path#

The event bus above is the substrate. Net's flagship use is agents discovering and invoking work across the mesh — a different loop on the same foundation, and the one the SDK is built for:

• Rust SDK quickstart — the same first program on the net_sdk surface, then a two-node Mesh that discovers capabilities. Start here if you skipped the crate table above.
• Discover and Invoke — query the mesh by capability (net-mesh cap query --tag …) and make a typed call.
• Wrap an MCP Server — turn an existing MCP tool into a discoverable capability with one command.
• Expose Net as MCP — let any MCP host use the mesh.
• The Agentic Mesh — the worldview behind all of it.