Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,23 @@ Run a local MLX model on your Apple Silicon Mac with
[Eredu](https://github.com/jbg/eredu), served through Mesh's normal
OpenAI-compatible API.

**Early access:** Apple Silicon (macOS arm64) only. Text chat and streaming;
tool calls are rejected for now.
**Early access:** Apple Silicon (macOS arm64) only. Text chat, streaming and
checkpoint-native function calls through OpenAI JSON/SSE. Tools require an Eredu
registered native-tool template; unsupported templates fail closed.

Tool requests support `tools`, `tool_choice` (`auto`, `none`, `required`),
`parallel_tool_calls` (default false), and `enable_thinking`. Named-function
`tool_choice` objects are not supported and are rejected, not silently ignored.
The plugin returns calls for your client to execute; it never executes tools.
Replay the assistant message and a `role: "tool"` message with its `tool_call_id`
to supply results. Function arguments on the wire are JSON strings.

Validated locally with an existing `mlx-community/Qwen3-1.7B-4bit` checkpoint:
required tool call, result replay with `tool_choice: "none"` and a nonempty
follow-up, plus streaming tool-call output (`enable_thinking: false`). This is
single-node adapter evidence, not qualification of every model, parallel tool
execution, agent harness, or remote Mesh routing. Readiness does not advertise
blanket model tool capability.

## 1. Install

Expand Down
174 changes: 170 additions & 4 deletions src/http.rs
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,8 @@ async fn chat(State(state): State<Endpoint>, Json(request): Json<ChatRequest>) -
let mut terminal = false;
while let Some(event) = generation.events.recv().await {
let payload = match event {
Event::ToolStart { index, id: call_id, name } => chunk(&id, &model, json!({"tool_calls":[{"index":index,"id":call_id,"type":"function","function":{"name":name,"arguments":""}}]}), None),
Event::ToolArguments { index, fragment } => chunk(&id, &model, json!({"tool_calls":[{"index":index,"function":{"arguments":fragment}}]}), None),
Event::Text(text) => chunk(&id, &model, json!({"content":text}), None),
Event::Finished(reason) => { terminal = true; chunk(&id, &model, json!({}), Some(reason)) },
Event::Error(message) => { terminal = true; json!({"error":{"message":message}}) },
Expand All @@ -84,15 +86,49 @@ async fn chat(State(state): State<Endpoint>, Json(request): Json<ChatRequest>) -
return Sse::new(stream).into_response();
}
let mut text = String::new();
let mut calls: Vec<Value> = Vec::new();
let mut output_bytes = 0usize;
while let Some(event) = generation.events.recv().await {
match event {
Event::ToolStart { index, id, name } => {
if index != calls.len() || calls.len() >= 128 {
return error(StatusCode::BAD_GATEWAY, "invalid tool call index".into());
}
output_bytes += id.len() + name.len();
calls.push(
json!({"id":id,"type":"function","function":{"name":name,"arguments":""}}),
);
}
Event::ToolArguments { index, fragment } => {
let Some(call) = calls.get_mut(index) else {
return error(
StatusCode::BAD_GATEWAY,
"arguments without tool call".into(),
);
};
output_bytes += fragment.len();
let args = call["function"]["arguments"].as_str().unwrap().to_owned() + &fragment;
call["function"]["arguments"] = Value::String(args);
}
Event::Text(delta) => {
if text.len() + delta.len() > 1024 * 1024 { return error(StatusCode::BAD_GATEWAY, "output exceeds limit".into()); }
output_bytes += delta.len();
text.push_str(&delta);
}
Event::Finished(reason) => return Json(json!({"id":id,"object":"chat.completion","model":model,"choices":[{"index":0,"message":{"role":"assistant","content":text},"finish_reason":reason}]})).into_response(),
Event::Finished(reason) => {
let mut message = json!({"role":"assistant","content":text});
if !calls.is_empty() {
if text.is_empty() {
message["content"] = Value::Null;
}
message["tool_calls"] = json!(calls);
}
return Json(json!({"id":id,"object":"chat.completion","model":model,"choices":[{"index":0,"message":message,"finish_reason":reason}]})).into_response();
}
Event::Error(message) => return error(StatusCode::BAD_GATEWAY, message),
}
if output_bytes > 1024 * 1024 {
return error(StatusCode::BAD_GATEWAY, "output exceeds limit".into());
}
}
error(
StatusCode::BAD_GATEWAY,
Expand Down Expand Up @@ -123,9 +159,9 @@ mod tests {
_: ChatRequest,
_: &ActiveWork,
output: &tokio::sync::mpsc::Sender<Event>,
) -> Result<()> {
) -> Result<&'static str> {
publish(output, Event::Text("héllo\nworld".into()))?;
publish(output, Event::Finished("stop"))
Ok("stop")
}
fn settle(&mut self) -> Result<()> {
Ok(())
Expand Down Expand Up @@ -163,4 +199,134 @@ mod tests {
}
assert!(frames[frames.len() - 2].contains("\"finish_reason\":\"stop\""));
}
struct ToolMock;
impl Backend for ToolMock {
fn generate(
&mut self,
request: ChatRequest,
_: &ActiveWork,
output: &tokio::sync::mpsc::Sender<Event>,
) -> Result<&'static str> {
if request.messages.iter().any(|m| m["role"] == "tool") {
assert_eq!(request.messages[1]["tool_calls"][0]["id"], "call_7");
assert_eq!(
request.messages[1]["tool_calls"][0]["function"]["arguments"],
json!({"city":"Bogotá"})
);
assert_eq!(request.messages[2]["tool_call_id"], "call_7");
publish(output, Event::Text("Sunny in Bogotá".into()))?;
return Ok("stop");
}
assert_eq!(request.tool_choice.as_deref(), Some("required"));
assert_eq!(request.tools[0]["function"]["name"], "weather");
publish(
output,
Event::ToolStart {
index: 0,
id: "call_7".into(),
name: "weather".into(),
},
)?;
publish(
output,
Event::ToolArguments {
index: 0,
fragment: "{\"city\":".into(),
},
)?;
publish(
output,
Event::ToolArguments {
index: 0,
fragment: "\"Bogotá\"}".into(),
},
)?;
Ok("tool_calls")
}
fn settle(&mut self) -> Result<()> {
Ok(())
}
}
async fn tool_call(body: Value) -> (StatusCode, String) {
let worker = Worker::start(|| Ok(ToolMock)).await.unwrap();
let app = router(Endpoint {
worker,
model: "local".into(),
});
let response = app
.oneshot(
Request::post("/v1/chat/completions")
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let bytes = to_bytes(response.into_body(), 65536).await.unwrap();
(status, String::from_utf8(bytes.to_vec()).unwrap())
}
fn tool_request(stream: bool) -> Value {
json!({"model":"local","stream":stream,"messages":[{"role":"user","content":"Weather?"}],
"tool_choice":"required","tools":[{"type":"function","function":{"name":"weather","parameters":{"type":"object"}}}]})
}
#[tokio::test]
async fn tool_json_and_result_replay_round_trip() {
let (status, body) = tool_call(tool_request(false)).await;
assert_eq!(status, StatusCode::OK);
let value: Value = serde_json::from_str(&body).unwrap();
let message = value["choices"][0]["message"].clone();
assert!(message["content"].is_null());
assert_eq!(value["choices"][0]["finish_reason"], "tool_calls");
assert_eq!(
message["tool_calls"][0]["function"]["arguments"],
"{\"city\":\"Bogotá\"}"
);
let (status, body) = tool_call(json!({"model":"local","messages":[
{"role":"user","content":"Weather?"}, message,
{"role":"tool","tool_call_id":"call_7","content":"sunny"}]}))
.await;
assert_eq!(status, StatusCode::OK);
let answer: Value = serde_json::from_str(&body).unwrap();
assert_eq!(
answer["choices"][0]["message"]["content"],
"Sunny in Bogotá"
);
}
#[tokio::test]
async fn tool_sse_fragments_reconstruct_openai_call() {
let (status, body) = tool_call(tool_request(true)).await;
assert_eq!(status, StatusCode::OK);
let mut arguments = String::new();
let mut id = String::new();
let mut name = String::new();
let mut finish = String::new();
for frame in body.split("\n\n").filter_map(|f| f.strip_prefix("data: ")) {
if frame == "[DONE]" {
continue;
}
let v: Value = serde_json::from_str(frame).unwrap();
let call = &v["choices"][0]["delta"]["tool_calls"][0];
if !call.is_null() {
assert_eq!(call["index"], 0);
if let Some(s) = call["id"].as_str() {
id.push_str(s);
}
if let Some(s) = call["function"]["name"].as_str() {
name.push_str(s);
}
if let Some(s) = call["function"]["arguments"].as_str() {
arguments.push_str(s);
}
}
if let Some(s) = v["choices"][0]["finish_reason"].as_str() {
finish = s.into();
}
}
assert_eq!(id, "call_7");
assert_eq!(name, "weather");
assert_eq!(arguments, "{\"city\":\"Bogotá\"}");
assert_eq!(finish, "tool_calls");
assert!(body.ends_with("data: [DONE]\n\n"));
}
}
60 changes: 50 additions & 10 deletions src/native.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ use eredu::{
LoadedModel, PreparedChatGenerationRequest, PreparedChatGenerationSettings,
PreparedChatInput, default_local_device, local_device_plan,
},
runtime::chat::ChatTemplateRequest,
runtime::chat::{ChatTemplateRequest, NativeToolSupport, ParallelToolCallPolicy, ToolChoice},
};
use eredu_backend_mlx::MlxBackendFactory;
use eredu_core::ExecutionPlanBackendFactory;
Expand Down Expand Up @@ -37,21 +37,38 @@ impl Backend for Native {
request: ChatRequest,
work: &ActiveWork,
output: &mpsc::Sender<Event>,
) -> Result<()> {
if !request.tools.is_empty()
) -> Result<&'static str> {
let needs_tools = !request.tools.is_empty()
|| request
.messages
.iter()
.any(|m| m.get("tool_calls").is_some() || m["role"] == "tool")
{
bail!("native tool calls are not yet validated by this adapter");
}
.any(|m| m.get("tool_calls").is_some() || m["role"] == "tool");
self.0.reset()?;
let tool_choice = match request.tool_choice.as_deref() {
Some("none") => ToolChoice::None,
Some("required") => ToolChoice::Required,
_ => ToolChoice::Auto,
};
let prepared = self.0.prepare_chat(ChatTemplateRequest {
messages: request.messages,
tools: request.tools,
tool_choice,
parallel_tool_calls: if request.parallel_tool_calls {
ParallelToolCallPolicy::Enabled { max_calls: None }
} else {
ParallelToolCallPolicy::Disabled
},
enable_thinking: request.enable_thinking,
add_generation_prompt: true,
..Default::default()
})?;
if needs_tools
&& let NativeToolSupport::Unsupported { reason } = prepared.native_tool_support()
{
bail!("native tools unavailable for this template: {reason}");
}
let mut started_calls = 0usize;
let mut ended_calls = 0usize;
let cancellation = GenerationCancellationToken::new();
let mut delivery_failed = false;
let result = self
Expand All @@ -74,8 +91,27 @@ impl Backend for Native {
cancellation.cancel();
return;
}
if let SemanticEvent::TextDelta(text) = event
&& publish(output, Event::Text(text)).is_err()
let event = match event {
SemanticEvent::TextDelta(text) => Some(Event::Text(text)),
SemanticEvent::ToolCallStart { index, id, name } => {
started_calls += 1;
Some(Event::ToolStart { index, id, name })
}
SemanticEvent::ToolArgumentsDelta {
index,
json_fragment,
} => Some(Event::ToolArguments {
index,
fragment: json_fragment,
}),
SemanticEvent::ToolCallEnd => {
ended_calls += 1;
None
}
_ => None,
};
if let Some(event) = event
&& publish(output, event).is_err()
{
delivery_failed = true;
cancellation.cancel();
Expand All @@ -85,12 +121,16 @@ impl Backend for Native {
if delivery_failed || work.is_cancelled() {
bail!("generation cancelled by consumer");
}
if started_calls != ended_calls {
bail!("generation ended with incomplete tool calls");
}
let reason = match result.finish_reason {
FinishReason::MaxTokens => "length",
FinishReason::Cancelled => bail!("generation cancelled"),
_ if started_calls > 0 => "tool_calls",
_ => "stop",
};
publish(output, Event::Finished(reason))
Ok(reason)
}

fn settle(&mut self) -> Result<()> {
Expand Down
Loading
Loading