diff --git a/Cargo.lock b/Cargo.lock index 9fdcf64..3936205 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1681,6 +1681,7 @@ dependencies = [ "prometheus-http-client", "rand", "regex", + "reqwest", "serde", "serde_json", "thiserror", diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index 339d480..e8e884b 100644 --- a/signal-gateway/Cargo.toml +++ b/signal-gateway/Cargo.toml @@ -26,6 +26,7 @@ http-body-util = { workspace = true } humantime = { workspace = true } jsonrpsee = { workspace = true } regex = { workspace = true } +reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/signal-gateway/src/claude/mod.rs b/signal-gateway/src/claude/mod.rs new file mode 100644 index 0000000..d234179 --- /dev/null +++ b/signal-gateway/src/claude/mod.rs @@ -0,0 +1,279 @@ +//! Claude API integration for AI-powered responses with tool use support. + +mod tools; +pub use tools::{Tool, ToolExecutor}; + +use conf::Conf; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::path::PathBuf; +use tracing::info; + +/// The Anthropic API version header value. This is a stable version identifier, +/// not a date indicating when the API was released. Anthropic adds new features +/// through separate `anthropic-beta` headers rather than bumping this version. +/// The official Anthropic SDKs use this same value. +const ANTHROPIC_API_VERSION: &str = "2023-06-01"; + +/// Configuration for the Claude API integration. +#[derive(Clone, Conf, Debug)] +#[conf(serde)] +pub struct ClaudeConfig { + /// Path to file containing the Claude API key. + #[conf(long, env)] + pub api_key_file: PathBuf, + /// Path to file containing the system prompt. + #[conf(long, env)] + pub system_prompt_file: PathBuf, + /// Claude API URL. + #[conf(long, env, default_value = "https://api.anthropic.com/v1/messages")] + pub claude_api_url: String, + /// Claude model to use. + #[conf(long, env, default_value = "claude-opus-4-20250514")] + pub claude_model: String, + /// Maximum tokens in the response. + #[conf(long, env, default_value = "1024")] + pub claude_max_tokens: u32, + /// Maximum tool use iterations before giving up. + #[conf(long, env, default_value = "10")] + pub claude_max_iterations: u32, +} + +/// Error type for Claude API operations. +#[derive(Debug, thiserror::Error)] +pub enum ClaudeError { + /// Failed to read API key file. + #[error("failed to read API key file: {0}")] + ApiKeyRead(std::io::Error), + /// Failed to read system prompt file. + #[error("failed to read system prompt file: {0}")] + SystemPromptRead(std::io::Error), + /// HTTP request failed. + #[error("HTTP request failed: {0}")] + Request(#[from] reqwest::Error), + /// API returned an error response. + #[error("API error: {0}")] + ApiError(String), + /// Tool execution failed. + #[error("tool execution failed: {0}")] + ToolError(String), + /// Too many tool use iterations. + #[error("exceeded maximum tool use iterations ({0})")] + TooManyIterations(u32), +} + +/// Claude API client. +pub struct ClaudeApi { + config: ClaudeConfig, + client: reqwest::Client, + api_key: String, + system_prompt: String, +} + +/// Request body for the Claude Messages API. +#[derive(Serialize)] +struct MessagesRequest<'a> { + model: &'a str, + max_tokens: u32, + system: &'a str, + messages: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + tools: Vec, +} + +/// A message in the conversation (can have multiple content blocks). +#[derive(Clone, Debug, Serialize, Deserialize)] +struct MessageContent { + role: String, + content: Vec, +} + +impl MessageContent { + fn user(text: &str) -> Self { + Self { + role: "user".to_owned(), + content: vec![ContentBlock::Text { + text: text.to_owned(), + }], + } + } + + fn assistant(blocks: Vec) -> Self { + Self { + role: "assistant".to_owned(), + content: blocks, + } + } + + fn tool_result(tool_use_id: String, content: String, is_error: bool) -> Self { + Self { + role: "user".to_owned(), + content: vec![ContentBlock::ToolResult { + tool_use_id, + content, + is_error: if is_error { Some(true) } else { None }, + }], + } + } +} + +/// A content block in the request/response. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum ContentBlock { + Text { + text: String, + }, + ToolUse { + id: String, + name: String, + input: Value, + }, + ToolResult { + tool_use_id: String, + content: String, + #[serde(skip_serializing_if = "Option::is_none")] + is_error: Option, + }, +} + +/// Response from the Claude Messages API. +#[derive(Debug, Deserialize)] +struct MessagesResponse { + content: Vec, + stop_reason: String, +} + +/// Error response from the Claude API. +#[derive(Deserialize)] +struct ErrorResponse { + error: ApiErrorDetail, +} + +#[derive(Deserialize)] +struct ApiErrorDetail { + message: String, +} + +impl ClaudeApi { + /// Create a new Claude API client from configuration. + /// + /// Reads the API key and system prompt from the configured files. + pub fn new(config: ClaudeConfig) -> Result { + let api_key = std::fs::read_to_string(&config.api_key_file) + .map_err(ClaudeError::ApiKeyRead)? + .trim() + .to_owned(); + + let system_prompt = std::fs::read_to_string(&config.system_prompt_file) + .map_err(ClaudeError::SystemPromptRead)?; + + let client = reqwest::Client::new(); + + Ok(Self { + config, + client, + api_key, + system_prompt, + }) + } + + /// Send a request to the Claude API and return the response text. + /// If a tool executor is provided, handles tool use in a loop. + pub async fn request( + &self, + prompt: &str, + tool_executor: Option<&dyn ToolExecutor>, + ) -> Result { + let max_iterations = self.config.claude_max_iterations; + + let tools = tool_executor.map(|te| te.tools()).unwrap_or_default(); + let mut messages = vec![MessageContent::user(prompt)]; + + info!("Claude request: {}", prompt); + + for iteration in 0..max_iterations { + let request_body = MessagesRequest { + model: &self.config.claude_model, + max_tokens: self.config.claude_max_tokens, + system: &self.system_prompt, + messages: messages.clone(), + tools: tools.clone(), + }; + + let response = self + .client + .post(&self.config.claude_api_url) + .header("x-api-key", &self.api_key) + .header("anthropic-version", ANTHROPIC_API_VERSION) + .header("content-type", "application/json") + .json(&request_body) + .send() + .await?; + + if !response.status().is_success() { + let error: ErrorResponse = response.json().await?; + return Err(ClaudeError::ApiError(error.error.message)); + } + + let response: MessagesResponse = response.json().await?; + info!( + "Claude response (stop_reason={}): {:?}", + response.stop_reason, response.content + ); + + // Check if we need to handle tool use + if response.stop_reason == "tool_use" { + let Some(executor) = tool_executor else { + return Err(ClaudeError::ToolError( + "tool use requested but no executor provided".to_owned(), + )); + }; + + // Add assistant's response to messages + messages.push(MessageContent::assistant(response.content.clone())); + + // Execute each tool use and collect results + for block in &response.content { + if let ContentBlock::ToolUse { id, name, input } = block { + info!("Claude tool use: {}({})", name, input); + let (result, is_error) = match executor.execute(name, input).await { + Ok(result) => { + info!("Tool result: {}", result); + (result, false) + } + Err(err) => { + info!("Tool error: {}", err); + (err, true) + } + }; + messages.push(MessageContent::tool_result(id.clone(), result, is_error)); + } + } + + // Continue the loop to get Claude's next response + info!("Tool use iteration {}, continuing...", iteration + 1); + continue; + } + + // No tool use, extract final text response + let text = response + .content + .into_iter() + .filter_map(|block| { + if let ContentBlock::Text { text } = block { + Some(text) + } else { + None + } + }) + .collect::>() + .join("\n"); + + info!("Claude final result: {}", text); + return Ok(text); + } + + Err(ClaudeError::TooManyIterations(max_iterations)) + } +} diff --git a/signal-gateway/src/claude/tools.rs b/signal-gateway/src/claude/tools.rs new file mode 100644 index 0000000..6d577dc --- /dev/null +++ b/signal-gateway/src/claude/tools.rs @@ -0,0 +1,32 @@ +//! Tool definitions and executor trait for the Claude API. + +use async_trait::async_trait; +use serde::Serialize; +use serde_json::Value; + +/// Trait for executing tools. Implement this to provide tool capabilities. +#[async_trait] +pub trait ToolExecutor: Send + Sync { + /// Get the list of available tools as JSON tool definitions. + fn tools(&self) -> Vec; + + /// Check if this executor handles a tool with the given name. + fn has_tool(&self, name: &str) -> bool { + self.tools().iter().any(|t| t.name == name) + } + + /// Execute a tool by name with the given input arguments. + /// Returns the result as a string to be sent back to Claude. + async fn execute(&self, name: &str, input: &Value) -> Result; +} + +/// A tool definition for the Claude API. +#[derive(Clone, Debug, Serialize)] +pub struct Tool { + /// The name of the tool. + pub name: &'static str, + /// A description of what the tool does. + pub description: &'static str, + /// JSON schema for the tool's input parameters. + pub input_schema: Value, +} diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index 40830c4..bacdd73 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/signal-gateway/src/gateway/log_handler.rs @@ -4,10 +4,12 @@ use super::{ route::{Destination, Limit, Route}, }; use crate::{ + claude::{Tool, ToolExecutor}, concurrent_map::LazyMap, log_format::LogFormatConfig, log_message::{LogFilter, LogMessage, Origin}, }; +use async_trait::async_trait; use chrono::Utc; use conf::Conf; use std::fmt; @@ -282,3 +284,37 @@ impl LogHandler { Ok(first_destination) } } + +fn logs_tool() -> Tool { + Tool { + name: "logs", + description: "Get recent log messages from monitored applications. Returns buffered log entries, optionally filtered by application name or hostname.", + input_schema: serde_json::json!({ + "type": "object", + "properties": { + "filter": { + "type": "string", + "description": "Optional filter string. If it contains '@', format is 'app@host' where both parts are substring matches. Otherwise, matches either app or host containing the string." + } + }, + "required": [] + }), + } +} + +#[async_trait] +impl ToolExecutor for LogHandler { + fn tools(&self) -> Vec { + vec![logs_tool()] + } + + async fn execute(&self, name: &str, input: &serde_json::Value) -> Result { + match name { + "logs" => { + let filter = input.get("filter").and_then(|v| v.as_str()); + Ok(self.format_logs(filter).await) + } + _ => Err(format!("unknown tool: {name}")), + } + } +} diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 4aa5db4..a36ec0f 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -4,6 +4,7 @@ use crate::signal_jsonrpc::connect_ipc; use crate::{ alertmanager::AlertPost, + claude::{ClaudeApi, ClaudeConfig, Tool, ToolExecutor}, log_message::{LogMessage, Origin}, message_handler::{ AdminMessage, AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, @@ -13,6 +14,7 @@ use crate::{ Envelope, MessageTarget, RpcClient, RpcClientError, SignalMessage, connect_tcp, }, }; +use async_trait::async_trait; use chrono::Utc; use conf::{Conf, Subcommands}; use futures_util::FutureExt; @@ -75,6 +77,9 @@ pub struct GatewayConfig { /// Log handler configuration for processing log messages. #[conf(flatten, prefix)] pub log_handler: LogHandlerConfig, + /// Claude API configuration for AI-powered responses. + #[conf(flatten, prefix)] + pub claude: Option, } /// Wrapper for parsing gateway commands @@ -129,6 +134,13 @@ enum GatewayCommand { /// Show current alerts from prometheus #[conf(name = "alerts", alias = "ALERTS")] Alerts, + /// Ask Claude AI a question + #[conf(name = "c", alias = "C")] + Claude { + /// The prompt to send to Claude + #[conf(repeat, pos)] + prompt: Vec, + }, } /// Parse a gateway command from a string (with or without leading /) @@ -217,6 +229,8 @@ pub struct Gateway { log_handler: LogHandler, /// Handler for admin messages that don't start with `/` message_handler: Option>, + /// Claude API client for AI-powered responses. + claude: Option, } impl Gateway { @@ -237,6 +251,11 @@ impl Gateway { let log_handler = LogHandler::new(config.log_handler.clone(), signal_alert_mq_tx.clone()); + let claude = config + .claude + .clone() + .map(|cc| ClaudeApi::new(cc).expect("Invalid claude config")); + Self { config, signal_alert_mq_tx, @@ -245,6 +264,7 @@ impl Gateway { prometheus, log_handler, message_handler, + claude, } } @@ -655,6 +675,19 @@ impl Gateway { Err(err) => Err((500, err)), } } + GatewayCommand::Claude { prompt } => { + let claude = self + .claude + .as_ref() + .ok_or_else(|| (501u16, "claude was not configured".into()))?; + + let prompt_text = prompt.join(" "); + // Pass self as the tool executor so Claude can query prometheus + match claude.request(&prompt_text, Some(self)).await { + Ok(response) => Ok(AdminMessageResponse::new(response)), + Err(err) => Err((500, err.to_string().into())), + } + } } } @@ -773,6 +806,31 @@ impl Gateway { } } +#[async_trait] +impl ToolExecutor for Gateway { + fn tools(&self) -> Vec { + let mut tools = self.log_handler.tools(); + if let Some(prometheus) = &self.prometheus { + tools.extend(prometheus.tools()); + } + tools + } + + async fn execute(&self, name: &str, input: &serde_json::Value) -> Result { + if self.log_handler.has_tool(name) { + return self.log_handler.execute(name, input).await; + } + + if let Some(prometheus) = &self.prometheus + && prometheus.has_tool(name) + { + return prometheus.execute(name, input).await; + } + + Err(format!("unknown tool: {name}")) + } +} + /// Placeholder context for message handlers. struct GatewayContext; diff --git a/signal-gateway/src/lib.rs b/signal-gateway/src/lib.rs index 8bbc3ea..7b35aae 100644 --- a/signal-gateway/src/lib.rs +++ b/signal-gateway/src/lib.rs @@ -6,6 +6,7 @@ #![deny(missing_docs)] pub mod alertmanager; +pub mod claude; pub mod gateway; pub mod message_handler; diff --git a/signal-gateway/src/log_message.rs b/signal-gateway/src/log_message.rs index 6cd3284..23194df 100644 --- a/signal-gateway/src/log_message.rs +++ b/signal-gateway/src/log_message.rs @@ -358,10 +358,10 @@ impl LogFilter { return false; } - if let Some(regex) = &self.msg_regex { - if !regex.is_match(&log_msg.msg) { - return false; - } + if let Some(regex) = &self.msg_regex + && !regex.is_match(&log_msg.msg) + { + return false; } if !self.module_equals.is_empty() { diff --git a/signal-gateway/src/prometheus/mod.rs b/signal-gateway/src/prometheus/mod.rs index a4f6e0c..1245604 100644 --- a/signal-gateway/src/prometheus/mod.rs +++ b/signal-gateway/src/prometheus/mod.rs @@ -1,9 +1,12 @@ +use crate::claude::{Tool, ToolExecutor}; +use async_trait::async_trait; use conf::Conf; use prometheus_http_client::{ AlertInfo, AlertsRequest, ExtractLabels, Labels, LabelsRequest, MetricVal, MetricValue, PromRequest, QueryRequest, ReqwestClient, SeriesRequest, }; use std::error::Error; +use std::fmt::Write; use tracing::info; type BoxError = Box; @@ -228,3 +231,172 @@ fn parse_alert_expr(expr: &str) -> Result<(String, PlotThreshold), BoxError> { Err("no comparator (< or >) found in expression".into()) } + +// Tool definitions for Claude API integration + +fn query_tool() -> Tool { + Tool { + name: "prometheus_query", + description: "Query Prometheus for current metric values using PromQL. Returns the current value of metrics matching the query expression.", + input_schema: serde_json::json!({ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "A PromQL query expression (e.g., 'up', 'rate(http_requests_total[5m])', 'node_memory_MemFree_bytes / node_memory_MemTotal_bytes')" + } + }, + "required": ["query"] + }), + } +} + +fn series_tool() -> Tool { + Tool { + name: "prometheus_series", + description: "List all time series (metrics) matching the given label matchers. Returns metric names with all their label key-value pairs.", + input_schema: serde_json::json!({ + "type": "object", + "properties": { + "matchers": { + "type": "array", + "items": { "type": "string" }, + "description": "Label matchers to filter series (e.g., ['__name__=~\"http_.*\"', 'job=\"api\"']). Use __name__ to match metric names." + } + }, + "required": ["matchers"] + }), + } +} + +fn labels_tool() -> Tool { + Tool { + name: "prometheus_labels", + description: "List all label names that exist in Prometheus, optionally filtered by series matchers.", + input_schema: serde_json::json!({ + "type": "object", + "properties": { + "matchers": { + "type": "array", + "items": { "type": "string" }, + "description": "Optional label matchers to filter which series to consider (e.g., ['job=\"api\"']). If empty, returns all label names." + } + }, + "required": [] + }), + } +} + +fn alerts_tool() -> Tool { + Tool { + name: "prometheus_alerts", + description: "List all current alerts from Prometheus, including their state (pending/firing), labels, and annotations.", + input_schema: serde_json::json!({ + "type": "object", + "properties": {}, + "required": [] + }), + } +} + +#[async_trait] +impl ToolExecutor for Prometheus { + fn tools(&self) -> Vec { + vec![query_tool(), series_tool(), labels_tool(), alerts_tool()] + } + + async fn execute(&self, name: &str, input: &serde_json::Value) -> Result { + match name { + "prometheus_query" => { + let query = input + .get("query") + .and_then(|v| v.as_str()) + .ok_or_else(|| "missing 'query' parameter".to_owned())?; + + match self.oneoff_query(query.to_owned()).await { + Ok((labels, values)) => { + let mut result = format!( + "Query: {}\nMetric: {} (common labels: {:?})\n", + query, labels.name, labels.common_labels + ); + for (sl, val) in labels.specific_labels.iter().zip(values.iter()) { + let value_str = val + .as_ref() + .map(|(_, v)| v.to_string()) + .unwrap_or_else(|| "-".to_owned()); + writeln!(&mut result, " {:?} = {}", sl, value_str).unwrap(); + } + Ok(result) + } + Err(err) => Err(format!("prometheus query failed: {err}")), + } + } + "prometheus_series" => { + let matchers: Vec = input + .get("matchers") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(|s| s.to_owned())) + .collect() + }) + .unwrap_or_default(); + + match self.series(&matchers).await { + Ok(series) => { + let mut result = format!("Found {} series:\n", series.len()); + for labels in series.iter().take(100) { + writeln!(&mut result, " {:?}", labels).unwrap(); + } + if series.len() > 100 { + result.push_str(&format!(" ... and {} more\n", series.len() - 100)); + } + Ok(result) + } + Err(err) => Err(format!("prometheus series failed: {err}")), + } + } + "prometheus_labels" => { + let matchers: Vec = input + .get("matchers") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(|s| s.to_owned())) + .collect() + }) + .unwrap_or_default(); + + match self.labels(&matchers).await { + Ok(labels) => { + let mut result = format!("Found {} labels:\n", labels.len()); + for label in &labels { + writeln!(&mut result, " {}", label).unwrap(); + } + Ok(result) + } + Err(err) => Err(format!("prometheus labels failed: {err}")), + } + } + "prometheus_alerts" => match self.alerts().await { + Ok(alerts) => { + if alerts.is_empty() { + return Ok("No active alerts".to_owned()); + } + let mut result = format!("Found {} alerts:\n", alerts.len()); + for alert in &alerts { + writeln!( + &mut result, + " [{:?}] {:?} - {:?}", + alert.state, alert.labels, alert.annotations + ) + .unwrap(); + } + Ok(result) + } + Err(err) => Err(format!("prometheus alerts failed: {err}")), + }, + _ => Err(format!("unknown tool: {name}")), + } + } +}