MCP | Ax

These Rust examples are real runnable files. Edit the source file first; this page is rebuilt from the checked-in example and its metadata header.

Rust Native MCP Tools

Attaches a live MCP client directly to AxGen without a lossy function adapter.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, MCP_URL
Level: beginner
Run: npm run example -- rust src/examples/rust/mcp/native_mcp_tools.rs
Source: src/examples/rust/mcp/native_mcp_tools.rs

Rust

use axllm::{
    ax, AxExecutionContext, AxMCPClient, AxMCPStreamableHTTPTransport, AxResult,
};
use serde_json::json;
use std::{
    env,
    sync::{Arc, Mutex},
};

fn main() -> AxResult<()> {
    let key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| axllm::AxError::runtime("Set OPENAI_API_KEY."))?;
    let endpoint = env::var("MCP_URL").map_err(|_| axllm::AxError::runtime("Set MCP_URL."))?;
    // The repo's demo MCP server runs on http://127.0.0.1; any other endpoint
    // keeps the default SSRF protection (https only, no local hosts).
    let local = endpoint.starts_with("http://127.0.0.1");
    let transport = AxMCPStreamableHTTPTransport::new(
        endpoint,
        json!({"ssrfProtection": {"requireHttps": !local, "allowLocalhost": local, "allowPrivateNetworks": local}}),
    )?;
    let mcp = Arc::new(Mutex::new(AxMCPClient::new(
        Box::new(transport),
        json!({"namespace":"inventory"}),
    )));
    let context = AxExecutionContext::new(vec![mcp.clone()], vec![])?;
    let catalog = mcp.lock().unwrap().inspect_catalog(false)?;
    println!(
        "MCP catalog: {} tools, {} resources, {} templates",
        catalog.tools.len(),
        catalog.resources.len(),
        catalog.resource_templates.len()
    );
    let mut program = ax("request:string -> answer:string")?.with_execution_context(context)?;
    let mut llm = axllm::ai("openai", json!({"api_key": key, "model": "gpt-5.4-mini"}))?;
    println!(
        "{}",
        program.forward(&mut llm, json!({"request":"Reindex inventory."}))?
    );
    mcp.lock().unwrap().close()?;
    Ok(())
}

Rust MCP Resource Wake

Subscribes over real Streamable HTTP and lets AxEventRuntime wake an authenticated Agent automatically.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, AX_MCP_ENDPOINT
Level: intermediate
Run: npm run example -- rust src/examples/rust/mcp/resource_wake_agent.rs
Source: src/examples/rust/mcp/resource_wake_agent.rs

Rust

use axllm::runtime::quickjs::QuickJsCodeRuntime;
use axllm::{
    agent_with_options, AxEventRoute, AxEventRuntime, AxEventTarget, AxMCPClient, AxMCPEventSource,
    AxMCPResourceSubscriptionPolicy, AxMCPStreamableHTTPTransport, AxResult,
};
use serde_json::{json, Value};
use std::{
    env,
    sync::{Arc, Condvar, Mutex},
    time::{Duration, Instant},
};

fn main() -> AxResult<()> {
    let key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| axllm::AxError::runtime("Set OPENAI_API_KEY."))?;
    let endpoint = env::var("AX_MCP_ENDPOINT").map_err(|_| {
        axllm::AxError::runtime("Set AX_MCP_ENDPOINT to a Streamable HTTP MCP server.")
    })?;
    let local = endpoint.starts_with("http://127.0.0.1");
    let transport = AxMCPStreamableHTTPTransport::new(
        endpoint.clone(),
        json!({"ssrfProtection":{"requireHttps":!local,"allowLocalhost":local,"allowPrivateNetworks":local}}),
    )?;
    let client = Arc::new(Mutex::new(AxMCPClient::new(
        Box::new(transport),
        json!({"namespace":"inventory"}),
    )));
    let mut llm = axllm::ai("openai", json!({"api_key": key, "model": "gpt-5.4-mini"}))?;
    let mut agent = agent_with_options(
        "uri:string -> summary:string",
        json!({"runtime":{"language":"JavaScript"}}),
    )?
    .with_runtime(Box::new(QuickJsCodeRuntime::new()))?;
    let completed = Arc::new((Mutex::new(false), Condvar::new()));
    let completed_target = completed.clone();
    let mut target = AxEventTarget::new("inventory-agent", move |input, _| {
        let output = agent.forward(&mut llm, input)?;
        println!("{output}");
        let (lock, changed) = &*completed_target;
        *lock.lock().unwrap() = true;
        changed.notify_all();
        Ok(output)
    });
    target.retry_safety = "idempotent".into();
    target.map_input = Some(Arc::new(|event, _| Ok(json!({"uri":event.data["uri"]}))));
    let mut runtime = AxEventRuntime::new(
        vec![AxEventRoute {
            id: "resource-wake".into(),
            action: "wake".into(),
            r#match: json!({"types":["mcp.resource.updated"]}),
            target_id: Some("inventory-agent".into()),
            require_authenticated: true,
            ordering: "strict".into(),
            debounce_ms: 0,
            instance_key: None,
        }],
        json!({}),
    )?;
    runtime.register_target(target);
    runtime.start()?;
    let runtime = Arc::new(Mutex::new(runtime));
    let mut source = AxMCPEventSource::with_policy(
        client.clone(),
        runtime.clone(),
        "inventory",
        "tenant:demo",
        "authenticated",
        AxMCPResourceSubscriptionPolicy::All,
    );
    source.start()?;
    if env::var("AX_MCP_DEMO_AUTO").as_deref() == Ok("1") {
        update_demo_resource(&endpoint)?;
    }
    let deadline = Instant::now() + Duration::from_secs(60);
    loop {
        source.poll();
        if *completed.0.lock().unwrap() {
            break;
        }
        if Instant::now() >= deadline {
            return Err(axllm::AxError::runtime(
                "Timed out waiting for an MCP resource notification",
            ));
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    source.close()?;
    client.lock().unwrap().close()?;
    runtime.lock().unwrap().close()?;
    Ok(())
}

// AX_MCP_DEMO_AUTO=1: the repo's demo server (plain HTTP on 127.0.0.1)
// updates the resource on request.
fn update_demo_resource(endpoint: &str) -> AxResult<()> {
    use std::io::{Read, Write};
    let authority = endpoint
        .trim_start_matches("http://")
        .split('/')
        .next()
        .unwrap_or_default();
    let failed = |error: std::io::Error| axllm::AxError::runtime(error.to_string());
    let mut stream = std::net::TcpStream::connect(authority).map_err(failed)?;
    write!(
        stream,
        "POST /control/resource HTTP/1.1\r\nHost: {authority}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
    )
    .map_err(failed)?;
    let mut response = String::new();
    stream.read_to_string(&mut response).map_err(failed)?;
    Ok(())
}

Rust MCP Task Continuation

Creates an owned continuation and resumes an AxFlow from real MCP progress and terminal task notifications.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, AX_MCP_ENDPOINT
Level: advanced
Run: npm run example -- rust src/examples/rust/mcp/task_resume_flow.rs
Source: src/examples/rust/mcp/task_resume_flow.rs

Rust

use axllm::{
    ax, AxEventEnvelope, AxEventRoute, AxEventRuntime, AxEventTarget, AxMCPClient,
    AxMCPEventSource, AxMCPStreamableHTTPTransport, AxMCPToolCallOutcome, AxResult,
};
use serde_json::{json, Map};
use std::{
    env,
    sync::{Arc, Condvar, Mutex},
    time::{Duration, Instant},
};

fn main() -> AxResult<()> {
    let key = env::var("OPENAI_API_KEY")
        .or_else(|_| env::var("OPENAI_APIKEY"))
        .map_err(|_| axllm::AxError::runtime("Set OPENAI_API_KEY."))?;
    let endpoint = env::var("AX_MCP_ENDPOINT").map_err(|_| {
        axllm::AxError::runtime("Set AX_MCP_ENDPOINT to a Streamable HTTP MCP server.")
    })?;
    let local = endpoint.starts_with("http://127.0.0.1");
    let transport = AxMCPStreamableHTTPTransport::new(
        endpoint.clone(),
        json!({"ssrfProtection":{"requireHttps":!local,"allowLocalhost":local,"allowPrivateNetworks":local}}),
    )?;
    let client = Arc::new(Mutex::new(AxMCPClient::new(
        Box::new(transport),
        json!({"namespace":"inventory"}),
    )));
    client.lock().unwrap().init()?;
    // A modern server answers with a task; a legacy server returns it inside a
    // complete result.
    let outcome = client
        .lock()
        .unwrap()
        .call_tool_outcome("start_reindex", json!({"scope":"all"}))?;
    let task = match outcome {
        AxMCPToolCallOutcome::Task(task) => Some(task),
        AxMCPToolCallOutcome::Complete(result) => result.get("task").cloned(),
    }
    .expect("start_reindex did not start an MCP task");
    let task_id = task["taskId"].as_str().unwrap().to_string();
    let mut llm = axllm::ai("openai", json!({"api_key": key, "model": "gpt-5.4-mini"}))?;
    let mut flow = axllm::flow("reindex-flow")
        .execute("status", ax("taskId:string -> status:string")?)
        .returns(json!({"status":"status"}));
    let completed = Arc::new((Mutex::new(0usize), Condvar::new()));
    let completed_target = completed.clone();
    let mut target = AxEventTarget::new("reindex-flow", move |input, _| {
        let output = flow.forward(&mut llm, input)?;
        for key in ["status"] {
            let value = &output[key];
            assert!(!value.is_null() && value != "" && *value != json!([]), "flow output field {key} is empty: {output}");
        }
        println!("{output}");
        let (lock, changed) = &*completed_target;
        *lock.lock().unwrap() += 1;
        changed.notify_all();
        Ok(output)
    });
    target.retry_safety = "idempotent".into();
    target.wait_for =
        vec![json!({"kind":"mcp.task","value":"taskKey","metadata":{"taskId":task_id}})];
    target.map_input = Some(Arc::new(|event, continuation| {
        Ok(
            json!({"taskId":continuation.map(|value|value.metadata["taskId"].clone()).unwrap_or_else(||event.data["taskId"].clone())}),
        )
    }));
    let routes = vec![
        AxEventRoute {
            id: "task-start".into(),
            action: "wake".into(),
            r#match: json!({"types":["app.task.started"]}),
            target_id: Some("reindex-flow".into()),
            require_authenticated: false,
            ordering: "strict".into(),
            debounce_ms: 0,
            instance_key: None,
        },
        AxEventRoute {
            id: "task-progress".into(),
            action: "observe".into(),
            r#match: json!({"types":["mcp.progress"]}),
            target_id: None,
            require_authenticated: false,
            ordering: "strict".into(),
            debounce_ms: 0,
            instance_key: None,
        },
        AxEventRoute {
            id: "task-resume".into(),
            action: "resume".into(),
            r#match: json!({"types":["mcp.task.status"]}),
            target_id: Some("reindex-flow".into()),
            require_authenticated: false,
            ordering: "strict".into(),
            debounce_ms: 0,
            instance_key: None,
        },
    ];
    let mut runtime_value = AxEventRuntime::new(routes, json!({}))?;
    runtime_value.register_target(target);
    runtime_value.start()?;
    runtime_value.publish(
        AxEventEnvelope {
            specversion: "1.0".into(),
            id: "task-start".into(),
            source: "app://tasks".into(),
            r#type: "app.task.started".into(),
            subject: Some(task_id.clone()),
            data: json!({"taskId":task_id,"taskKey":format!("inventory:{task_id}")}),
            extensions: Map::new(),
            correlation: vec![],
        },
        "tenant:demo",
        "authenticated",
    )?;
    let runtime = Arc::new(Mutex::new(runtime_value));
    let mut source = AxMCPEventSource::new(
        client.clone(),
        runtime.clone(),
        "inventory",
        "tenant:demo",
        "authenticated",
        vec![],
    );
    source.start()?;
    println!("Waiting for terminal MCP task notification {task_id}");
    if env::var("AX_MCP_DEMO_AUTO").as_deref() == Ok("1") {
        complete_demo_task(&endpoint)?;
    }
    let deadline = Instant::now() + Duration::from_secs(60);
    loop {
        source.poll();
        if *completed.0.lock().unwrap() >= 2 {
            break;
        }
        if Instant::now() >= deadline {
            return Err(axllm::AxError::runtime(
                "Timed out waiting for the MCP task continuation",
            ));
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    source.close()?;
    client.lock().unwrap().close()?;
    runtime.lock().unwrap().close()?;
    Ok(())
}

// AX_MCP_DEMO_AUTO=1: the repo's demo server (plain HTTP on 127.0.0.1)
// completes the task on request.
fn complete_demo_task(endpoint: &str) -> AxResult<()> {
    use std::io::{Read, Write};
    let authority = endpoint
        .trim_start_matches("http://")
        .split('/')
        .next()
        .unwrap_or_default();
    let failed = |error: std::io::Error| axllm::AxError::runtime(error.to_string());
    let mut stream = std::net::TcpStream::connect(authority).map_err(failed)?;
    write!(
        stream,
        "POST /control/task/complete HTTP/1.1\r\nHost: {authority}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
    )
    .map_err(failed)?;
    let mut response = String::new();
    stream.read_to_string(&mut response).map_err(failed)?;
    Ok(())
}