1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::consul::{
8 actions::{action, parameter},
9 wire,
10};
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Context, Result};
13use serde_json::{json, Value};
14use std::sync::LazyLock;
15
16#[derive(Default)]
17pub struct ConsulClientProtocol;
18impl ConsulClientProtocol {
19 pub fn new() -> Self {
20 Self
21 }
22}
23
24fn get_action() -> ActionDefinition {
25 action(
26 "consul_kv_get",
27 "Read a key (or every key under a prefix with recurse, or only key names with keys_only).",
28 vec![
29 parameter("key", "string", "The key or prefix, e.g. app/config", true),
30 parameter(
31 "recurse",
32 "boolean",
33 "Read every entry under the prefix",
34 false,
35 ),
36 parameter(
37 "keys_only",
38 "boolean",
39 "List only the key names under the prefix",
40 false,
41 ),
42 ],
43 json!({"type":"consul_kv_get","key":"app/config"}),
44 "-> Consul GET kv/{key}",
45 )
46}
47
48fn put_action() -> ActionDefinition {
49 action(
50 "consul_kv_put",
51 "Write a key; with cas, only if its modify index matches (0: only if absent).",
52 vec![
53 parameter("key", "string", "The key to write", true),
54 parameter(
55 "value",
56 "string",
57 "The value, as text (or hex with encoding hex)",
58 true,
59 ),
60 parameter("encoding", "string", "utf8 (default) or hex", false),
61 parameter(
62 "flags",
63 "number",
64 "Opaque flags stored with the value",
65 false,
66 ),
67 parameter("cas", "number", "Check-and-set modify index", false),
68 ],
69 json!({"type":"consul_kv_put","key":"app/config","value":"hello"}),
70 "-> Consul PUT kv/{key}",
71 )
72}
73
74fn delete_action() -> ActionDefinition {
75 action(
76 "consul_kv_delete",
77 "Delete a key, or everything under a prefix with recurse.",
78 vec![
79 parameter("key", "string", "The key or prefix to delete", true),
80 parameter(
81 "recurse",
82 "boolean",
83 "Delete everything under the prefix",
84 false,
85 ),
86 ],
87 json!({"type":"consul_kv_delete","key":"app/config"}),
88 "-> Consul DELETE kv/{key}",
89 )
90}
91
92fn catalog_action() -> ActionDefinition {
93 action(
94 "consul_catalog",
95 "Ask about services: services (all names and tags), service or health (one service's instances), nodes, agent_services.",
96 vec![
97 parameter("endpoint", "string", "services, service, health, nodes or agent_services", true),
98 parameter("name", "string", "The service, for service and health", false),
99 ],
100 json!({"type":"consul_catalog","endpoint":"services"}),
101 "-> Consul catalog {endpoint}",
102 )
103}
104
105fn register_action() -> ActionDefinition {
106 action(
107 "consul_register_service",
108 "Register a service with the agent.",
109 vec![
110 parameter(
111 "name",
112 "string",
113 "The name the service is registered under",
114 true,
115 ),
116 parameter("id", "string", "Service ID (defaults to the name)", false),
117 parameter(
118 "address",
119 "string",
120 "The address the service listens on",
121 false,
122 ),
123 parameter("port", "number", "The port the service listens on", false),
124 parameter(
125 "tags",
126 "array",
127 "Tags to attach to the service, e.g. [\"v1\"]",
128 false,
129 ),
130 ],
131 json!({"type":"consul_register_service","name":"web","port":8080,"tags":["v1"]}),
132 "-> Consul register {name}",
133 )
134}
135
136fn deregister_action() -> ActionDefinition {
137 action(
138 "consul_deregister_service",
139 "Remove a service registration from the agent.",
140 vec![parameter("id", "string", "The service ID to remove", true)],
141 json!({"type":"consul_deregister_service","id":"web"}),
142 "-> Consul deregister {id}",
143 )
144}
145
146fn disconnect_action() -> ActionDefinition {
147 action(
148 "disconnect",
149 "End this Consul client session.",
150 vec![],
151 json!({"type":"disconnect"}),
152 "-> Consul disconnect",
153 )
154}
155
156pub fn actions() -> Vec<ActionDefinition> {
157 vec![
158 get_action(),
159 put_action(),
160 delete_action(),
161 catalog_action(),
162 register_action(),
163 deregister_action(),
164 disconnect_action(),
165 ]
166}
167
168pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
169 EventType::new(
170 "consul_connected",
171 "The client is ready; each request opens its own HTTP connection.",
172 get_action().example.clone(),
173 )
174 .with_parameters(vec![parameter(
175 "remote_addr",
176 "string",
177 "The Consul agent's HTTP origin",
178 true,
179 )])
180 .with_actions(actions())
181});
182
183pub static RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
184 EventType::new("consul_response", "The agent answered a request.", catalog_action().example.clone())
185 .with_parameters(vec![
186 parameter("operation", "string", "The action that was performed", true),
187 parameter("status", "number", "The HTTP status the agent answered with", true),
188 parameter("result", "any", "KV entries {key, value, encoding, flags, modify_index}, key names, true/false, services or instances", true),
189 parameter("index", "number", "X-Consul-Index of the answer", true),
190 parameter("message", "string", "The agent's error text, when the status is not 200", true),
191 ])
192 .with_actions(actions())
193});
194
195impl Protocol for ConsulClientProtocol {
196 fn protocol_name(&self) -> &'static str {
197 "Consul"
198 }
199 fn stack_name(&self) -> &'static str {
200 "ETH>IP>TCP>HTTP>Consul"
201 }
202 fn keywords(&self) -> Vec<&'static str> {
203 vec!["consul", "consul kv", "service discovery"]
204 }
205 fn description(&self) -> &'static str {
206 "Consul HTTP API client: KV, catalog, health and service registration"
207 }
208 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
209 actions()
210 }
211 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
212 vec![]
213 }
214 fn get_event_types(&self) -> Vec<EventType> {
215 vec![CONNECTED_EVENT.clone(), RESPONSE_EVENT.clone()]
216 }
217 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
218 vec![]
219 }
220 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
221 use crate::protocol::metadata::*;
222 ProtocolMetadataV2::builder()
223 .state(DevelopmentState::Experimental)
224 .privilege_requirement(PrivilegeRequirement::None)
225 .implementation("HTTP/1.1 via hyper, one connection per request: /v1/kv get/put/delete, catalog, health, agent service register/deregister; base64 values decoded for the handler")
226 .llm_control("Which keys to read and write, which services to look up or register, and what to do with each answer")
227 .e2e_testing("tests/client/consul: NetGet's own agent; the official consul agent (-dev), read back with the consul CLI")
228 .notes("No blocking queries, sessions, ACL tokens or TLS. Answers are capped at 1 MiB. A handler chain stops after 8 follow-ups.")
229 .build()
230 }
231 fn example_prompt(&self) -> &'static str {
232 "Write app/config=hello to the Consul agent at 127.0.0.1:8500 and list its services"
233 }
234 fn get_startup_examples(&self) -> StartupExamples {
235 let llm = json!({"type":"open_client","protocol":"consul","remote_addr":"127.0.0.1:8500","instruction":"Store app/config=hello, then read it back"});
236 let mut static_example = llm.clone();
237 static_example["event_handlers"] = json!([{"event_pattern":"consul_connected","handler":{"type":"static","actions":[{"type":"consul_kv_put","key":"app/config","value":"hello"}]}}]);
238 let mut scripted = llm.clone();
239 scripted["event_handlers"] = json!([{"event_pattern":"consul_response","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\na=[{'type':'consul_kv_get','key':'app/config'}] if e['operation']=='consul_kv_put' else []\nprint(json.dumps({'actions':a}))"}}]);
240 StartupExamples::new(llm, scripted, static_example)
241 }
242 fn group_name(&self) -> &'static str {
243 "Network Services"
244 }
245}
246
247pub fn request(v: &Value) -> Result<(&'static str, String, Option<Vec<u8>>)> {
249 let enc = |s: &str| urlencoding::encode(s).into_owned().replace("%2F", "/");
250 let key = || -> Result<String> {
251 let k = v["key"].as_str().context("key required")?;
252 wire::check_key(k)?;
253 ensure!(
254 !k.is_empty() || v["recurse"] == true || v["keys_only"] == true,
255 "key required"
256 );
257 Ok(enc(k))
258 };
259 Ok(match v["type"].as_str().unwrap_or_default() {
260 "consul_kv_get" => {
261 let mut p = format!("/v1/kv/{}", key()?);
262 if v["keys_only"] == true {
263 p.push_str("?keys");
264 } else if v["recurse"] == true {
265 p.push_str("?recurse");
266 }
267 ("GET", p, None)
268 }
269 "consul_kv_put" => {
270 let mut q = Vec::new();
271 if let Some(f) = v["flags"].as_u64() {
272 q.push(format!("flags={f}"));
273 }
274 if let Some(c) = v["cas"].as_u64() {
275 q.push(format!("cas={c}"));
276 }
277 let mut p = format!("/v1/kv/{}", key()?);
278 if !q.is_empty() {
279 p.push('?');
280 p.push_str(&q.join("&"));
281 }
282 ("PUT", p, Some(wire::value_bytes(v)?))
283 }
284 "consul_kv_delete" => {
285 let mut p = format!("/v1/kv/{}", key()?);
286 if v["recurse"] == true {
287 p.push_str("?recurse");
288 }
289 ("DELETE", p, None)
290 }
291 "consul_catalog" => {
292 let name = || -> Result<String> {
293 let n = v["name"]
294 .as_str()
295 .context("name required for this endpoint")?;
296 ensure!(!n.is_empty(), "name required");
297 Ok(enc(n))
298 };
299 let p = match v["endpoint"].as_str().unwrap_or_default() {
300 "services" => "/v1/catalog/services".to_string(),
301 "nodes" => "/v1/catalog/nodes".to_string(),
302 "agent_services" => "/v1/agent/services".to_string(),
303 "service" => format!("/v1/catalog/service/{}", name()?),
304 "health" => format!("/v1/health/service/{}", name()?),
305 e => bail!("unknown endpoint {e}"),
306 };
307 ("GET", p, None)
308 }
309 "consul_register_service" => {
310 let i = wire::instance(v)?;
311 let body = json!({"ID": i.id, "Name": i.name, "Address": i.address, "Port": i.port, "Tags": i.tags});
312 (
313 "PUT",
314 "/v1/agent/service/register".into(),
315 Some(body.to_string().into_bytes()),
316 )
317 }
318 "consul_deregister_service" => {
319 let id = v["id"].as_str().context("id required")?;
320 ensure!(!id.is_empty(), "id required");
321 (
322 "PUT",
323 format!("/v1/agent/service/deregister/{}", enc(id)),
324 None,
325 )
326 }
327 t => bail!("Unknown Consul client action {t}"),
328 })
329}
330
331impl Client for ConsulClientProtocol {
332 fn connect(
333 &self,
334 ctx: ConnectContext,
335 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
336 {
337 Box::pin(super::connect(ctx))
338 }
339 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
340 let name = v["type"].as_str().unwrap_or_default().to_string();
341 if name == "disconnect" {
342 return Ok(ClientActionResult::Disconnect);
343 }
344 request(&v)?;
345 Ok(ClientActionResult::Custom { name, data: v })
346 }
347}