diff --git a/Cargo.toml b/Cargo.toml index 9d97a0e..9f88819 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,14 @@ [workspace] resolver = "2" -members = ["core/agentsd", "core/proto"] +members = [ + "core/agentsd", + "core/proto", + "plugins/plugin-sdk", + "plugins/echo", + "plugins/ai_test", + "plugins/hermes", + "plugins/claudecode", +] [workspace.dependencies] tokio = { version = "1", features = ["full"] } @@ -16,3 +24,4 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } anyhow = "1" uuid = { version = "1", features = ["v4"] } notify = "7" +futures-core = "0.3" diff --git a/core/agentsd/src/plugin.rs b/core/agentsd/src/plugin.rs index c4e4b90..a4304e9 100644 --- a/core/agentsd/src/plugin.rs +++ b/core/agentsd/src/plugin.rs @@ -16,6 +16,8 @@ pub struct PluginRecord { pub syscalls: Vec, pub depends: Vec, pub exports: Vec<(String, String)>, // (name, description) + #[serde(default)] + pub endpoint: Option, pub status: PluginStatus, } @@ -95,6 +97,11 @@ impl PluginManager { None } + pub fn find_export_record(&self, target: &str, fn_name: &str) -> Option { + let plugin_id = self.find_export(target, fn_name)?; + self.get(&plugin_id) + } + pub fn unregister(&self, id: &str) -> Result<(), String> { let mut plugins = self.plugins.lock().unwrap(); plugins diff --git a/plugins/ai_test/Cargo.toml b/plugins/ai_test/Cargo.toml new file mode 100644 index 0000000..e2f6625 --- /dev/null +++ b/plugins/ai_test/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "agentsd-plugin-ai-test" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "agentsd-plugin-ai-test" +path = "src/main.rs" + +[dependencies] +agentsd-plugin-sdk = { path = "../plugin-sdk" } +agentsd-proto = { path = "../../core/proto" } +anyhow = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } +tokio-stream = { workspace = true } +tonic = { workspace = true } diff --git a/plugins/ai_test/src/main.rs b/plugins/ai_test/src/main.rs new file mode 100644 index 0000000..066e5dc --- /dev/null +++ b/plugins/ai_test/src/main.rs @@ -0,0 +1,509 @@ +use agentsd_plugin_sdk::{export, plugin_info, request_with_plugin}; +use agentsd_proto::agentsd::{ + display_client::DisplayClient, fs_client::FsClient, memory_client::MemoryClient, + network_client::NetworkClient, plugin_bridge_client::PluginBridgeClient, + process_client::ProcessClient, timer_client::TimerClient, CallRequest, DisplayTextRequest, + Empty, FsPathRequest, FsReadRequest, FsRenameRequest, FsWriteRequest, MemoryKeyRequest, + MemoryListRequest, MemoryReadRequest, MemoryWriteRequest, PidRequest, SpawnRequest, + TimerOnceRequest, +}; +use anyhow::{anyhow, bail, ensure, Context, Result}; +use serde::Serialize; +use serde_json::Value; +use std::collections::HashMap; +use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::time::{timeout, Duration}; +use tokio_stream::StreamExt; +use tonic::transport::Channel; +use tonic::Code; + +const PLUGIN_ID: &str = "ai-test"; +const PROCESS_OK: &str = "AI_TEST_PROCESS_OK"; +const TIMER_CALLBACK_ID: &str = "ai-test-callback"; + +#[derive(Debug, Serialize)] +struct CheckResult { + name: String, + ok: bool, + details: String, +} + +#[derive(Debug, Serialize)] +struct Summary<'a> { + plugin: &'a str, + ok: bool, + results: &'a [CheckResult], +} + +struct CheckRunner { + channel: Channel, + results: Vec, + run_id: String, + scratch_dir: PathBuf, +} + +impl CheckRunner { + fn new(channel: Channel) -> Self { + let run_id = now_millis().to_string(); + let scratch_dir = executable_data_dir().join("ai_test").join(&run_id); + + Self { + channel, + results: Vec::new(), + run_id, + scratch_dir, + } + } + + async fn run(&mut self) -> i32 { + println!( + "AI Test Plugin connecting to {}", + agentsd_plugin_sdk::agentsd_addr() + ); + println!("Scratch dir: {}", self.scratch_dir.display()); + + let result = self.register().await; + self.record_result("register", result); + let result = self.display_text().await; + self.record_result("display.text", result); + let result = self.memory_roundtrip().await; + self.record_result("memory roundtrip", result); + let result = self.fs_roundtrip().await; + self.record_result("fs roundtrip", result); + let result = self.process_roundtrip().await; + self.record_result("process roundtrip", result); + let result = self.timer_event().await; + self.record_result("timer event", result); + let result = self.permission_denial().await; + self.record_result("permission denial", result); + + let ok = self.results.iter().all(|r| r.ok); + let summary = Summary { + plugin: PLUGIN_ID, + ok, + results: &self.results, + }; + let summary_json = serde_json::to_string(&summary) + .unwrap_or_else(|e| format!(r#"{{"plugin":"{PLUGIN_ID}","ok":false,"error":"{e}"}}"#)); + println!("AI_TEST_SUMMARY {summary_json}"); + + if ok { + 0 + } else { + 1 + } + } + + fn record_result(&mut self, name: &str, result: Result) { + match result { + Ok(details) => { + self.record(name, true, if details.is_empty() { "ok" } else { &details }) + } + Err(err) => self.record(name, false, &err.to_string()), + } + } + + fn record(&mut self, name: &str, ok: bool, details: &str) { + let status = if ok { "PASS" } else { "FAIL" }; + println!("[{status}] {name}: {details}"); + self.results.push(CheckResult { + name: name.to_string(), + ok, + details: details.to_string(), + }); + } + + async fn register(&mut self) -> Result { + let mut bridge = PluginBridgeClient::new(self.channel.clone()); + let resp = bridge + .register(plugin_info( + PLUGIN_ID, + "AI Test Plugin", + "0.1.0", + "basic", + &[ + "display.text", + "memory.write", + "memory.read", + "memory.list", + "memory.delete", + "fs.write", + "fs.read", + "fs.stat", + "fs.list", + "fs.rename", + "fs.delete", + "process.spawn", + "process.stdout", + "process.wait", + "timer.once", + ], + &[], + vec![export("run_checks", "Run Agentsd verification checks")], + )) + .await + .context("register RPC")? + .into_inner(); + + ensure!(resp.ok, "{}", resp.error); + Ok(format!("session={}", resp.session_id)) + } + + async fn display_text(&mut self) -> Result { + let mut display = DisplayClient::new(self.channel.clone()); + let resp = display + .text(request_with_plugin( + PLUGIN_ID, + DisplayTextRequest { + content: format!("[ai-test] display check run={}", self.run_id), + }, + )?) + .await + .context("display.text RPC")? + .into_inner(); + + ensure!(resp.ok, "{}", resp.error); + Ok("display.text ok".to_string()) + } + + async fn memory_roundtrip(&mut self) -> Result { + let mut memory = MemoryClient::new(self.channel.clone()); + let key = format!("ai_test.{}.greeting", self.run_id); + let value = format!("hello-{}", self.run_id).into_bytes(); + + let write = memory + .write(request_with_plugin( + PLUGIN_ID, + MemoryWriteRequest { + key: key.clone(), + value: value.clone(), + }, + )?) + .await + .context("memory.write RPC")? + .into_inner(); + ensure!(write.ok, "write: {}", write.error); + + let read = memory + .read(request_with_plugin( + PLUGIN_ID, + MemoryReadRequest { key: key.clone() }, + )?) + .await + .context("memory.read RPC")? + .into_inner(); + ensure!( + read.found && read.value == value, + "read mismatch found={} value={:?} error={:?}", + read.found, + read.value, + read.error + ); + + let list = memory + .list(request_with_plugin( + PLUGIN_ID, + MemoryListRequest { + prefix: format!("ai_test.{}", self.run_id), + limit: 10, + }, + )?) + .await + .context("memory.list RPC")? + .into_inner(); + ensure!( + list.error.is_empty() && list.keys.iter().any(|k| k == &key), + "list missing key; keys={:?} error={:?}", + list.keys, + list.error + ); + + let delete = memory + .delete(request_with_plugin( + PLUGIN_ID, + MemoryKeyRequest { key: key.clone() }, + )?) + .await + .context("memory.delete RPC")? + .into_inner(); + ensure!(delete.ok, "delete: {}", delete.error); + + Ok(format!("key={key}")) + } + + async fn fs_roundtrip(&mut self) -> Result { + let mut fs = FsClient::new(self.channel.clone()); + let base = self.scratch_dir.clone(); + let file_a = base.join("sample.txt"); + let file_b = base.join("renamed.txt"); + let data = format!("AI_TEST_FS_OK {}", self.run_id).into_bytes(); + + let write = fs + .write(request_with_plugin( + PLUGIN_ID, + FsWriteRequest { + path: path_string(&file_a), + data: data.clone(), + append: false, + }, + )?) + .await + .context("fs.write RPC")? + .into_inner(); + ensure!(write.ok, "write: {}", write.error); + + let read = fs + .read(request_with_plugin( + PLUGIN_ID, + FsReadRequest { + path: path_string(&file_a), + offset: 0, + length: 0, + }, + )?) + .await + .context("fs.read RPC")? + .into_inner(); + ensure!( + read.ok && read.data == data, + "read mismatch ok={} data={:?} error={:?}", + read.ok, + read.data, + read.error + ); + + let stat = fs + .stat(request_with_plugin( + PLUGIN_ID, + FsPathRequest { + path: path_string(&file_a), + }, + )?) + .await + .context("fs.stat RPC")? + .into_inner(); + ensure!( + stat.ok && stat.is_file && stat.size == data.len() as i64, + "stat mismatch ok={} is_file={} size={} error={:?}", + stat.ok, + stat.is_file, + stat.size, + stat.error + ); + + let list = fs + .list(request_with_plugin( + PLUGIN_ID, + FsPathRequest { + path: path_string(&base), + }, + )?) + .await + .context("fs.list RPC")? + .into_inner(); + let names: Vec<&str> = list.entries.iter().map(|e| e.name.as_str()).collect(); + ensure!( + list.ok && names.iter().any(|name| *name == "sample.txt"), + "list mismatch ok={} names={:?} error={:?}", + list.ok, + names, + list.error + ); + + let rename = fs + .rename(request_with_plugin( + PLUGIN_ID, + FsRenameRequest { + from: path_string(&file_a), + to: path_string(&file_b), + }, + )?) + .await + .context("fs.rename RPC")? + .into_inner(); + ensure!(rename.ok, "rename: {}", rename.error); + + let delete = fs + .delete(request_with_plugin( + PLUGIN_ID, + FsPathRequest { + path: path_string(&base), + }, + )?) + .await + .context("fs.delete RPC")? + .into_inner(); + ensure!(delete.ok, "delete: {}", delete.error); + + Ok(path_string(&base)) + } + + async fn process_roundtrip(&mut self) -> Result { + let mut process = ProcessClient::new(self.channel.clone()); + let resp = process + .spawn(request_with_plugin( + PLUGIN_ID, + SpawnRequest { + cmd: "cmd".to_string(), + args: vec!["/C".to_string(), format!("echo {PROCESS_OK}")], + env: HashMap::new(), + cwd: executable_dir() + .map(|path| path_string(&path)) + .unwrap_or_default(), + }, + )?) + .await + .context("process.spawn RPC")? + .into_inner(); + ensure!(resp.ok, "{}", resp.error); + let pid = resp.pid; + + let mut stdout = process + .stdout(request_with_plugin(PLUGIN_ID, PidRequest { pid })?) + .await + .context("process.stdout RPC")? + .into_inner(); + let mut chunks = Vec::new(); + while let Some(chunk) = stdout.next().await { + chunks.extend(chunk.context("process.stdout stream")?.data); + } + let output = String::from_utf8_lossy(&chunks).trim().to_string(); + + let wait = process + .wait(request_with_plugin(PLUGIN_ID, PidRequest { pid })?) + .await + .context("process.wait RPC")? + .into_inner(); + ensure!( + wait.ok && wait.exit_code == 0, + "wait failed ok={} code={} error={:?}", + wait.ok, + wait.exit_code, + wait.error + ); + ensure!(output == PROCESS_OK, "stdout mismatch: {output:?}"); + + Ok(format!("pid={pid} stdout={output}")) + } + + async fn timer_event(&mut self) -> Result { + let mut bridge = PluginBridgeClient::new(self.channel.clone()); + let mut timer = TimerClient::new(self.channel.clone()); + + let mut stream = bridge + .stream_call(request_with_plugin( + PLUGIN_ID, + CallRequest { + target: PLUGIN_ID.to_string(), + r#fn: "events".to_string(), + args: Vec::new(), + }, + )?) + .await + .context("plugin_bridge.stream_call RPC")? + .into_inner(); + + let once = timer + .once(request_with_plugin( + PLUGIN_ID, + TimerOnceRequest { + delay_ms: 100, + callback_id: TIMER_CALLBACK_ID.to_string(), + }, + )?) + .await + .context("timer.once RPC")? + .into_inner(); + ensure!(once.ok, "{}", once.error); + + let event = timeout(Duration::from_secs(5), stream.message()) + .await + .context("timed out waiting for timer event")? + .context("timer event stream")? + .ok_or_else(|| anyhow!("timer event stream ended before event"))?; + ensure!(event.ok, "stream event failed: {}", event.error); + + let payload: Value = + serde_json::from_slice(&event.result).context("decode timer event JSON")?; + ensure!( + payload.get("event").and_then(Value::as_str) == Some("timer") + && payload.get("callback_id").and_then(Value::as_str) == Some(TIMER_CALLBACK_ID), + "unexpected payload: {payload}" + ); + + Ok(serde_json::to_string(&payload)?) + } + + async fn permission_denial(&mut self) -> Result { + let mut network = NetworkClient::new(self.channel.clone()); + match network + .available(request_with_plugin(PLUGIN_ID, Empty {})?) + .await + { + Ok(_) => bail!("network.available unexpectedly succeeded"), + Err(status) if status.code() == Code::PermissionDenied => { + Ok(status.message().to_string()) + } + Err(status) => bail!("wrong error: {} {}", status.code(), status.message()), + } + } +} + +#[tokio::main] +async fn main() { + let code = match agentsd_plugin_sdk::connect().await { + Ok(channel) => { + let mut runner = CheckRunner::new(channel); + runner.run().await + } + Err(err) => { + let result = CheckResult { + name: "connect".to_string(), + ok: false, + details: err.to_string(), + }; + println!( + "AI Test Plugin connecting to {}", + agentsd_plugin_sdk::agentsd_addr() + ); + println!("[FAIL] connect: {}", result.details); + let results = vec![result]; + let summary = Summary { + plugin: PLUGIN_ID, + ok: false, + results: &results, + }; + let summary_json = serde_json::to_string(&summary).unwrap_or_else(|e| { + format!(r#"{{"plugin":"{PLUGIN_ID}","ok":false,"error":"{e}"}}"#) + }); + println!("AI_TEST_SUMMARY {summary_json}"); + 1 + } + }; + + std::process::exit(code); +} + +fn now_millis() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() +} + +fn executable_dir() -> Option { + std::env::current_exe() + .ok() + .and_then(|exe| exe.parent().map(|parent| parent.to_path_buf())) +} + +fn executable_data_dir() -> PathBuf { + executable_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join("data") +} + +fn path_string(path: &std::path::Path) -> String { + path.to_string_lossy().to_string() +} diff --git a/plugins/claudecode/Cargo.toml b/plugins/claudecode/Cargo.toml new file mode 100644 index 0000000..e926ca9 --- /dev/null +++ b/plugins/claudecode/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "agentsd-plugin-claudecode" +version = "2.1.169" +edition = "2021" + +[[bin]] +name = "agentsd-plugin-claudecode" +path = "src/main.rs" + +[dependencies] +agentsd-plugin-sdk = { path = "../plugin-sdk" } +agentsd-proto = { path = "../../core/proto" } +anyhow = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } +tonic = { workspace = true } +futures-core = { workspace = true } +tracing = { workspace = true } diff --git a/plugins/claudecode/src/main.rs b/plugins/claudecode/src/main.rs new file mode 100644 index 0000000..81455fa --- /dev/null +++ b/plugins/claudecode/src/main.rs @@ -0,0 +1,254 @@ +use agentsd_plugin_sdk::{decode_json, err_json, export, ok_json, plugin_info_with_endpoint}; +use agentsd_proto::agentsd::{ + plugin_bridge_server::{PluginBridge, PluginBridgeServer}, + CallRequest, CallResponse, HeartbeatRequest, HeartbeatResponse, PluginInfo, RegisterResponse, +}; +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::process::Stdio; +use tokio::io::AsyncWriteExt; +use tonic::{transport::Server, Request, Response, Status}; + +const PLUGIN_ID: &str = "claudecode"; +const DEFAULT_PORT: u16 = 50053; + +#[derive(Debug, Deserialize)] +struct CallArgs { + #[serde(default)] + prompt: String, + #[serde(default)] + text: String, + #[serde(default)] + cwd: String, +} + +#[derive(Debug, Serialize)] +struct CallResult { + result: String, +} + +fn find_claude_bin() -> Option { + if let Ok(bin) = std::env::var("CLAUDE_BIN") { + return Some(bin); + } + + let candidates = vec![ + "claude", + "claude.cmd", + &std::env::var("APPDATA") + .map(|p| format!("{}\\npm\\claude.cmd", p)) + .unwrap_or_default(), + ]; + + for candidate in candidates { + if std::process::Command::new(candidate) + .arg("--version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok() + { + return Some(candidate.to_string()); + } + } + None +} + +async fn call_claude( + claude_bin: &str, + prompt: &str, + cwd: Option<&str>, + max_turns: Option, + allowed_tools: &[&str], +) -> Result { + let mut cmd = tokio::process::Command::new(claude_bin); + cmd.arg("-p").arg("--output-format").arg("text"); + + if let Some(turns) = max_turns { + cmd.arg("--max-turns").arg(turns.to_string()); + } + + if !allowed_tools.is_empty() { + cmd.arg("--allowedTools"); + for tool in allowed_tools { + cmd.arg(tool); + } + } + + if let Some(dir) = cwd { + if !dir.is_empty() { + cmd.current_dir(dir); + } + } + + cmd.env("CLAUDE_CODE_SIMPLE", "1"); + cmd.stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let mut child = cmd.spawn().context("spawn claude process")?; + + if let Some(mut stdin) = child.stdin.take() { + stdin + .write_all(prompt.as_bytes()) + .await + .context("write prompt to stdin")?; + stdin.flush().await.context("flush stdin")?; + drop(stdin); + } + + let output = tokio::time::timeout(std::time::Duration::from_secs(300), child.wait_with_output()) + .await + .context("claude timeout (300s)")??; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + anyhow::bail!("claude failed: {}", stderr); + } + + Ok(String::from_utf8_lossy(&output.stdout).trim().to_string()) +} + +struct ClaudeCodePlugin { + claude_bin: String, +} + +impl ClaudeCodePlugin { + fn new(claude_bin: String) -> Self { + Self { claude_bin } + } + + async fn handle_call(&self, fn_name: &str, args: CallArgs) -> CallResponse { + let prompt = if !args.prompt.is_empty() { + args.prompt + } else { + args.text + }; + + if prompt.is_empty() { + return err_json("missing prompt or text"); + } + + let cwd = if args.cwd.is_empty() { + None + } else { + Some(args.cwd.as_str()) + }; + + let result = match fn_name { + "chat" => call_claude(&self.claude_bin, &prompt, cwd, Some(1), &[]).await, + "code" => { + call_claude( + &self.claude_bin, + &prompt, + cwd, + None, + &["Bash", "Read", "Edit", "Write", "Glob", "Grep"], + ) + .await + } + "review" => { + let review_prompt = + format!("Review the following code or changes. Be concise and actionable:\n\n{}", prompt); + call_claude( + &self.claude_bin, + &review_prompt, + cwd, + Some(1), + &["Read", "Glob", "Grep"], + ) + .await + } + "ask" => { + call_claude( + &self.claude_bin, + &prompt, + cwd, + Some(3), + &["Read", "Glob", "Grep"], + ) + .await + } + "run" => call_claude(&self.claude_bin, &prompt, cwd, None, &[]).await, + _ => { + return err_json(format!("unknown function: {}", fn_name)); + } + }; + + match result { + Ok(output) => ok_json(&CallResult { result: output }), + Err(e) => err_json(e.to_string()), + } + } +} + +#[tonic::async_trait] +impl PluginBridge for ClaudeCodePlugin { + async fn register(&self, _req: Request) -> Result, Status> { + Err(Status::unimplemented("not a bridge")) + } + + async fn call(&self, req: Request) -> Result, Status> { + let r = req.into_inner(); + let args: CallArgs = decode_json(&r.args).unwrap_or_default(); + let response = self.handle_call(&r.r#fn, args).await; + Ok(Response::new(response)) + } + + type StreamCallStream = futures_core::stream::Empty>; + + async fn stream_call(&self, _req: Request) -> Result, Status> { + Err(Status::unimplemented("stream_call not supported")) + } + + async fn heartbeat(&self, _req: Request) -> Result, Status> { + Ok(Response::new(HeartbeatResponse { ok: true })) + } +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt::init(); + + let claude_bin = find_claude_bin().context("Claude Code CLI not found. Install: npm install -g @anthropic-ai/claude-code")?; + tracing::info!("Found Claude Code CLI: {}", claude_bin); + + let endpoint = agentsd_plugin_sdk::plugin_endpoint_from_env(DEFAULT_PORT); + let listen_addr = agentsd_plugin_sdk::listen_addr_from_env(DEFAULT_PORT); + + let channel = agentsd_plugin_sdk::connect().await?; + let register_resp = agentsd_plugin_sdk::register( + channel, + plugin_info_with_endpoint( + PLUGIN_ID, + "Claude Code", + env!("CARGO_PKG_VERSION"), + "standard", + &["process.spawn", "display.text", "fs.read", "fs.write", "network.request"], + &["hermes"], + vec![ + export("chat", "Send prompt to Claude Code, get text response (no tools)"), + export("code", "Generate or edit code with Claude Code (Bash/Read/Edit/Write)"), + export("review", "Review code changes (Read/Glob/Grep only)"), + export("ask", "Ask about the codebase (Read/Glob/Grep, max 3 turns)"), + export("run", "Full Claude Code access (all tools, unlimited turns)"), + ], + &endpoint, + ), + ) + .await?; + + tracing::info!("Registered with agentsd: session={}", register_resp.session_id); + + let plugin = ClaudeCodePlugin::new(claude_bin); + let addr = listen_addr.parse()?; + + tracing::info!("Claude Code plugin server listening on {}", addr); + + Server::builder() + .add_service(PluginBridgeServer::new(plugin)) + .serve(addr) + .await?; + + Ok(()) +} diff --git a/plugins/echo/Cargo.toml b/plugins/echo/Cargo.toml new file mode 100644 index 0000000..ca7d386 --- /dev/null +++ b/plugins/echo/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "agentsd-plugin-echo" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "agentsd-plugin-echo" +path = "src/main.rs" + +[dependencies] +agentsd-plugin-sdk = { path = "../plugin-sdk" } +agentsd-proto = { path = "../../core/proto" } +anyhow = { workspace = true } +tokio = { workspace = true } +tonic = { workspace = true } diff --git a/plugins/echo/src/main.rs b/plugins/echo/src/main.rs new file mode 100644 index 0000000..1680f0e --- /dev/null +++ b/plugins/echo/src/main.rs @@ -0,0 +1,92 @@ +use agentsd_plugin_sdk::{export, plugin_info, request_with_plugin}; +use agentsd_proto::agentsd::{ + display_client::DisplayClient, fs_client::FsClient, memory_client::MemoryClient, + plugin_bridge_client::PluginBridgeClient, DisplayTextRequest, FsPathRequest, MemoryReadRequest, + MemoryWriteRequest, +}; +use anyhow::{ensure, Result}; + +const PLUGIN_ID: &str = "echo-test"; + +#[tokio::main] +async fn main() -> Result<()> { + let channel = agentsd_plugin_sdk::connect().await?; + + let mut bridge = PluginBridgeClient::new(channel.clone()); + let register = bridge + .register(plugin_info( + PLUGIN_ID, + "Echo Test Plugin", + "0.1.0", + "basic", + &["display.text", "fs.list", "memory.read", "memory.write"], + &[], + vec![export("echo", "Echo test")], + )) + .await? + .into_inner(); + ensure!(register.ok, "register failed: {}", register.error); + println!("Register: ok={} session={}", register.ok, register.session_id); + + let mut display = DisplayClient::new(channel.clone()); + let resp = display + .text(request_with_plugin( + PLUGIN_ID, + DisplayTextRequest { + content: "Hello from echo plugin!".to_string(), + }, + )?) + .await? + .into_inner(); + ensure!(resp.ok, "Display.Text failed: {}", resp.error); + println!("Display.Text: ok={}", resp.ok); + + let mut memory = MemoryClient::new(channel.clone()); + let resp = memory + .write(request_with_plugin( + PLUGIN_ID, + MemoryWriteRequest { + key: "test.greeting".to_string(), + value: b"Hello World".to_vec(), + }, + )?) + .await? + .into_inner(); + ensure!(resp.ok, "Memory.Write failed: {}", resp.error); + println!("Memory.Write: ok={}", resp.ok); + + let resp = memory + .read(request_with_plugin( + PLUGIN_ID, + MemoryReadRequest { + key: "test.greeting".to_string(), + }, + )?) + .await? + .into_inner(); + ensure!(resp.found, "Memory.Read missing value: {}", resp.error); + println!( + "Memory.Read: found={} value={}", + resp.found, + String::from_utf8_lossy(&resp.value) + ); + + let mut fs = FsClient::new(channel); + let resp = fs + .list(request_with_plugin( + PLUGIN_ID, + FsPathRequest { + path: ".".to_string(), + }, + )?) + .await? + .into_inner(); + ensure!(resp.ok, "FS.List failed: {}", resp.error); + println!("FS.List: ok={} entries={}", resp.ok, resp.entries.len()); + for entry in resp.entries.iter().take(5) { + println!(" {} {}", entry.name, if entry.is_dir { "[dir]" } else { "" }); + } + + println!("\nAll syscalls verified OK!"); + Ok(()) +} diff --git a/plugins/hermes/Cargo.toml b/plugins/hermes/Cargo.toml new file mode 100644 index 0000000..86e3554 --- /dev/null +++ b/plugins/hermes/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "agentsd-plugin-hermes" +version = "0.16.0" +edition = "2021" + +[[bin]] +name = "agentsd-plugin-hermes" +path = "src/main.rs" + +[dependencies] +agentsd-plugin-sdk = { path = "../plugin-sdk" } +agentsd-proto = { path = "../../core/proto" } +anyhow = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } +tokio-stream = { workspace = true, features = ["net"] } +tonic = { workspace = true } diff --git a/plugins/hermes/src/main.rs b/plugins/hermes/src/main.rs new file mode 100644 index 0000000..4a3a071 --- /dev/null +++ b/plugins/hermes/src/main.rs @@ -0,0 +1,117 @@ +use agentsd_plugin_sdk::{ + decode_json, err_json, export, listen_addr_from_env, ok_json, plugin_endpoint_from_env, + plugin_info_with_endpoint, register, +}; +use agentsd_proto::agentsd::{ + plugin_bridge_server::{PluginBridge, PluginBridgeServer}, + CallRequest, CallResponse, HeartbeatRequest, HeartbeatResponse, PluginInfo, RegisterResponse, +}; +use anyhow::{ensure, Result}; +use serde::Deserialize; +use serde_json::json; +use tokio::net::TcpListener; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::{transport::Server, Request, Response, Status}; + +const PLUGIN_ID: &str = "hermes"; +const PLUGIN_NAME: &str = "Hermes Agent Tools"; +const PLUGIN_VERSION: &str = "0.16.0"; +const PLUGIN_TYPE: &str = "basic"; +const PLUGIN_PORT: u16 = 50052; +const SYSCALLS: &[&str] = &[ + "network.request", + "fs.read", + "fs.write", + "process.spawn", + "memory.read", + "memory.write", + "display.text", + "timer.once", +]; + +#[derive(Default)] +struct HermesPlugin; + +#[derive(Deserialize)] +struct EchoArgs { + #[serde(default)] + text: String, +} + +#[tonic::async_trait] +impl PluginBridge for HermesPlugin { + async fn register( + &self, + _req: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("hermes plugin is not an agentsd bridge")) + } + + async fn call( + &self, + req: Request, + ) -> std::result::Result, Status> { + let req = req.into_inner(); + let response = match req.r#fn.as_str() { + "echo" => match decode_json::(&req.args) { + Ok(args) => ok_json(&json!({ "result": format!("hermes: {}", args.text) })), + Err(err) => err_json(format!("invalid echo args: {err}")), + }, + other => err_json(format!("unknown export: {other}")), + }; + Ok(Response::new(response)) + } + + type StreamCallStream = tokio_stream::wrappers::ReceiverStream>; + + async fn stream_call( + &self, + _req: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("stream_call is not implemented for hermes")) + } + + async fn heartbeat( + &self, + _req: Request, + ) -> std::result::Result, Status> { + Ok(Response::new(HeartbeatResponse { ok: true })) + } +} + +fn plugin_info(endpoint: &str) -> PluginInfo { + plugin_info_with_endpoint( + PLUGIN_ID, + PLUGIN_NAME, + PLUGIN_VERSION, + PLUGIN_TYPE, + SYSCALLS, + &[], + vec![export("echo", "Echo test route for Hermes plugin")], + endpoint, + ) +} + +#[tokio::main] +async fn main() -> Result<()> { + let endpoint = plugin_endpoint_from_env(PLUGIN_PORT); + let listen_addr: std::net::SocketAddr = listen_addr_from_env(PLUGIN_PORT).parse()?; + let listener = TcpListener::bind(listen_addr).await?; + let incoming = TcpListenerStream::new(listener); + + let channel = agentsd_plugin_sdk::connect().await?; + let resp = register(channel, plugin_info(&endpoint)).await?; + ensure!(resp.ok, "register failed: {}", resp.error); + + println!( + "Hermes plugin registered: id={} type={} endpoint={} session={}", + PLUGIN_ID, PLUGIN_TYPE, endpoint, resp.session_id + ); + println!("Hermes PluginBridge server listening on {}", listen_addr); + + Server::builder() + .add_service(PluginBridgeServer::new(HermesPlugin::default())) + .serve_with_incoming(incoming) + .await?; + Ok(()) +} diff --git a/plugins/plugin-sdk/Cargo.toml b/plugins/plugin-sdk/Cargo.toml new file mode 100644 index 0000000..d5889fd --- /dev/null +++ b/plugins/plugin-sdk/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "agentsd-plugin-sdk" +version = "0.1.0" +edition = "2021" + +[dependencies] +agentsd-proto = { path = "../../core/proto" } +anyhow = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tonic = { workspace = true } +tokio = { workspace = true } +tokio-stream = { workspace = true } +tracing = { workspace = true } diff --git a/plugins/plugin-sdk/src/lib.rs b/plugins/plugin-sdk/src/lib.rs new file mode 100644 index 0000000..729c911 --- /dev/null +++ b/plugins/plugin-sdk/src/lib.rs @@ -0,0 +1,132 @@ +use agentsd_proto::agentsd::*; +use anyhow::{anyhow, Context, Result}; +use serde::{de::DeserializeOwned, Serialize}; +use tonic::metadata::MetadataValue; +use tonic::transport::Channel; +use tonic::Request; + +pub const DEFAULT_AGENTSD_ADDR: &str = "http://[::1]:50051"; + +pub fn agentsd_addr() -> String { + std::env::var("AGENTSD_ADDR").unwrap_or_else(|_| DEFAULT_AGENTSD_ADDR.to_string()) +} + +pub async fn connect() -> Result { + connect_to(&agentsd_addr()).await +} + +pub async fn connect_to(addr: &str) -> Result { + Channel::from_shared(addr.to_string()) + .with_context(|| format!("invalid agentsd address: {addr}"))? + .connect() + .await + .with_context(|| format!("connect agentsd at {addr}")) +} + +pub fn request_with_plugin(plugin_id: &str, message: T) -> Result> { + let mut req = Request::new(message); + let value = MetadataValue::try_from(plugin_id) + .map_err(|_| anyhow!("invalid plugin id for metadata: {plugin_id}"))?; + req.metadata_mut().insert("x-plugin-id", value); + Ok(req) +} + +pub fn export(name: &str, description: &str) -> ExportDef { + ExportDef { + name: name.to_string(), + description: description.to_string(), + } +} + +pub fn plugin_info( + id: &str, + name: &str, + version: &str, + plugin_type: &str, + syscalls: &[&str], + depends: &[&str], + exports: Vec, +) -> PluginInfo { + PluginInfo { + id: id.to_string(), + name: name.to_string(), + version: version.to_string(), + plugin_type: plugin_type.to_string(), + syscalls: syscalls.iter().map(|s| s.to_string()).collect(), + depends: depends.iter().map(|s| s.to_string()).collect(), + exports, + endpoint: String::new(), + } +} + +pub fn plugin_info_with_endpoint( + id: &str, + name: &str, + version: &str, + plugin_type: &str, + syscalls: &[&str], + depends: &[&str], + exports: Vec, + endpoint: &str, +) -> PluginInfo { + let mut info = plugin_info(id, name, version, plugin_type, syscalls, depends, exports); + info.endpoint = endpoint.to_string(); + info +} + +pub async fn register(channel: Channel, info: PluginInfo) -> Result { + let mut bridge = plugin_bridge_client::PluginBridgeClient::new(channel); + let resp = bridge.register(info).await?.into_inner(); + if !resp.ok { + return Err(anyhow!(resp.error)); + } + Ok(resp) +} + +pub fn decode_json(bytes: &[u8]) -> Result { + if bytes.is_empty() { + serde_json::from_str("{}").context("decode empty JSON object") + } else { + serde_json::from_slice(bytes).context("decode JSON") + } +} + +pub fn encode_json(value: &T) -> Result> { + serde_json::to_vec(value).context("encode JSON") +} + +pub fn call_response_json(value: &T, ok: bool, error: impl Into) -> CallResponse { + match encode_json(value) { + Ok(result) => CallResponse { + result, + ok, + error: error.into(), + }, + Err(e) => CallResponse { + result: Vec::new(), + ok: false, + error: e.to_string(), + }, + } +} + +pub fn ok_json(value: &T) -> CallResponse { + call_response_json(value, true, "") +} + +pub fn err_json(error: impl Into) -> CallResponse { + let error = error.into(); + let payload = serde_json::json!({ "error": error }); + call_response_json(&payload, false, error) +} + +pub fn plugin_endpoint_from_env(default_port: u16) -> String { + if let Ok(endpoint) = std::env::var("PLUGIN_ENDPOINT") { + return endpoint; + } + format!("http://[::1]:{}", default_port) +} + +pub fn listen_addr_from_env(default_port: u16) -> String { + std::env::var("PLUGIN_LISTEN_ADDR").unwrap_or_else(|_| format!("[::1]:{}", default_port)) +} diff --git a/proto/agentsd.proto b/proto/agentsd.proto index 40f1558..9993760 100644 --- a/proto/agentsd.proto +++ b/proto/agentsd.proto @@ -205,6 +205,7 @@ message PluginInfo { repeated string syscalls = 5; repeated string depends = 6; repeated ExportDef exports = 7; + string endpoint = 8; } message ExportDef { string name = 1; string description = 2; } message RegisterResponse { bool ok = 1; string error = 2; string session_id = 3; }