feat(mcp): add LLM server crate and remote client integration
- Introduce `owlen-mcp-llm-server` crate with RPC handling, `generate_text` tool, model listing, and streaming notifications. - Add `RpcNotification` struct and `MODELS_LIST` method to the MCP protocol. - Update `owlen-core` to depend on `tokio-stream`. - Adjust Ollama provider to omit empty `tools` field for compatibility. - Enhance `RemoteMcpClient` to locate the renamed server binary, handle resource tools locally, and implement the `Provider` trait (model listing, chat, streaming, health check). - Add new crate to workspace `Cargo.toml`.
This commit is contained in:
@@ -161,6 +161,25 @@ impl RpcErrorResponse {
|
||||
}
|
||||
}
|
||||
|
||||
/// JSON‑RPC notification (no id). Used for streaming partial results.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RpcNotification {
|
||||
pub jsonrpc: String,
|
||||
pub method: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub params: Option<Value>,
|
||||
}
|
||||
|
||||
impl RpcNotification {
|
||||
pub fn new(method: impl Into<String>, params: Option<Value>) -> Self {
|
||||
Self {
|
||||
jsonrpc: JSONRPC_VERSION.to_string(),
|
||||
method: method.into(),
|
||||
params,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Request ID can be string, number, or null
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
|
||||
#[serde(untagged)]
|
||||
@@ -194,6 +213,7 @@ pub mod methods {
|
||||
pub const RESOURCES_GET: &str = "resources/get";
|
||||
pub const RESOURCES_WRITE: &str = "resources/write";
|
||||
pub const RESOURCES_DELETE: &str = "resources/delete";
|
||||
pub const MODELS_LIST: &str = "models/list";
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
|
||||
@@ -1,11 +1,19 @@
|
||||
use super::protocol::{RequestId, RpcErrorResponse, RpcRequest, RpcResponse};
|
||||
use super::protocol::methods;
|
||||
use super::protocol::{RequestId, RpcErrorResponse, RpcRequest, RpcResponse, PROTOCOL_VERSION};
|
||||
use super::{McpClient, McpToolCall, McpToolDescriptor, McpToolResponse};
|
||||
use crate::{Error, Result};
|
||||
use crate::types::ModelInfo;
|
||||
use crate::{Error, Provider, Result};
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::process::{Child, Command};
|
||||
use tokio::sync::Mutex;
|
||||
// Provider trait is already imported via the earlier use statement.
|
||||
use crate::types::{ChatResponse, Message, Role};
|
||||
use futures::stream;
|
||||
use futures::StreamExt;
|
||||
|
||||
/// Client that talks to the external `owlen-mcp-server` over STDIO.
|
||||
pub struct RemoteMcpClient {
|
||||
@@ -28,11 +36,32 @@ impl RemoteMcpClient {
|
||||
// Attempt to locate the server binary; if unavailable we will fall back to launching via `cargo run`.
|
||||
let _ = ();
|
||||
// Resolve absolute path based on workspace root to avoid cwd dependence.
|
||||
// The MCP server binary lives in the workspace's `target/debug` directory.
|
||||
// Historically the binary was named `owlen-mcp-server`, but it has been
|
||||
// renamed to `owlen-mcp-llm-server`. We attempt to locate the new name
|
||||
// first and fall back to the legacy name for compatibility.
|
||||
let workspace_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../..")
|
||||
.canonicalize()
|
||||
.map_err(Error::Io)?;
|
||||
let binary_path = workspace_root.join("target/debug/owlen-mcp-server");
|
||||
let candidates = [
|
||||
"target/debug/owlen-mcp-llm-server",
|
||||
"target/debug/owlen-mcp-server",
|
||||
];
|
||||
let mut binary_path = None;
|
||||
for rel in &candidates {
|
||||
let p = workspace_root.join(rel);
|
||||
if p.exists() {
|
||||
binary_path = Some(p);
|
||||
break;
|
||||
}
|
||||
}
|
||||
let binary_path = binary_path.ok_or_else(|| {
|
||||
Error::NotImplemented(format!(
|
||||
"owlen-mcp server binary not found; checked {} and {}",
|
||||
candidates[0], candidates[1]
|
||||
))
|
||||
})?;
|
||||
if !binary_path.exists() {
|
||||
return Err(Error::NotImplemented(format!(
|
||||
"owlen-mcp-server binary not found at {}",
|
||||
@@ -107,8 +136,48 @@ impl McpClient for RemoteMcpClient {
|
||||
}
|
||||
|
||||
async fn call_tool(&self, call: McpToolCall) -> Result<McpToolResponse> {
|
||||
let result = self.send_rpc(&call.name, call.arguments.clone()).await?;
|
||||
// The remote server returns only the tool result; we fabricate metadata.
|
||||
// Local handling for simple resource tools to avoid needing the MCP server
|
||||
// to implement them.
|
||||
if call.name.starts_with("resources/get") {
|
||||
let path = call
|
||||
.arguments
|
||||
.get("path")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("");
|
||||
let content = std::fs::read_to_string(path).map_err(Error::Io)?;
|
||||
return Ok(McpToolResponse {
|
||||
name: call.name,
|
||||
success: true,
|
||||
output: serde_json::json!(content),
|
||||
metadata: std::collections::HashMap::new(),
|
||||
duration_ms: 0,
|
||||
});
|
||||
}
|
||||
if call.name.starts_with("resources/list") {
|
||||
let path = call
|
||||
.arguments
|
||||
.get("path")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or(".");
|
||||
let mut names = Vec::new();
|
||||
for entry in std::fs::read_dir(path).map_err(Error::Io)?.flatten() {
|
||||
if let Some(name) = entry.file_name().to_str() {
|
||||
names.push(name.to_string());
|
||||
}
|
||||
}
|
||||
return Ok(McpToolResponse {
|
||||
name: call.name,
|
||||
success: true,
|
||||
output: serde_json::json!(names),
|
||||
metadata: std::collections::HashMap::new(),
|
||||
duration_ms: 0,
|
||||
});
|
||||
}
|
||||
// MCP server expects a generic "tools/call" method with a payload containing the
|
||||
// specific tool name and its arguments. Wrap the incoming call accordingly.
|
||||
let payload = serde_json::to_value(&call)?;
|
||||
let result = self.send_rpc(methods::TOOLS_CALL, payload).await?;
|
||||
// The server returns the tool's output directly; construct a matching response.
|
||||
Ok(McpToolResponse {
|
||||
name: call.name,
|
||||
success: true,
|
||||
@@ -118,3 +187,66 @@ impl McpClient for RemoteMcpClient {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Provider implementation – forwards chat requests to the generate_text tool.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[async_trait]
|
||||
impl Provider for RemoteMcpClient {
|
||||
fn name(&self) -> &str {
|
||||
"mcp-llm-server"
|
||||
}
|
||||
|
||||
async fn list_models(&self) -> Result<Vec<ModelInfo>> {
|
||||
let result = self.send_rpc(methods::MODELS_LIST, json!(null)).await?;
|
||||
let models: Vec<ModelInfo> = serde_json::from_value(result)?;
|
||||
Ok(models)
|
||||
}
|
||||
|
||||
async fn chat(&self, request: crate::types::ChatRequest) -> Result<ChatResponse> {
|
||||
// Use the streaming implementation and take the first response.
|
||||
let mut stream = self.chat_stream(request).await?;
|
||||
match stream.next().await {
|
||||
Some(Ok(resp)) => Ok(resp),
|
||||
Some(Err(e)) => Err(e),
|
||||
None => Err(Error::Provider(anyhow::anyhow!("Empty chat stream"))),
|
||||
}
|
||||
}
|
||||
|
||||
async fn chat_stream(
|
||||
&self,
|
||||
request: crate::types::ChatRequest,
|
||||
) -> Result<crate::provider::ChatStream> {
|
||||
// Build arguments matching the generate_text schema.
|
||||
let args = serde_json::json!({
|
||||
"messages": request.messages,
|
||||
"temperature": request.parameters.temperature,
|
||||
"max_tokens": request.parameters.max_tokens,
|
||||
"model": request.model,
|
||||
"stream": request.parameters.stream,
|
||||
});
|
||||
let call = McpToolCall {
|
||||
name: "generate_text".to_string(),
|
||||
arguments: args,
|
||||
};
|
||||
let resp = self.call_tool(call).await?;
|
||||
// Build a ChatResponse from the tool output (assumed to be a string).
|
||||
let content = resp.output.as_str().unwrap_or("").to_string();
|
||||
let message = Message::new(Role::Assistant, content);
|
||||
let chat_resp = ChatResponse {
|
||||
message,
|
||||
usage: None,
|
||||
is_streaming: false,
|
||||
is_final: true,
|
||||
};
|
||||
let stream = stream::once(async move { Ok(chat_resp) });
|
||||
Ok(Box::pin(stream))
|
||||
}
|
||||
|
||||
async fn health_check(&self) -> Result<()> {
|
||||
// Simple ping using initialize method.
|
||||
let params = serde_json::json!({"protocol_version": PROTOCOL_VERSION});
|
||||
self.send_rpc("initialize", params).await.map(|_| ())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user