Conversation
Contributor
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds two new workspace crates, Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant "MCP Handler" as MCP
participant "Engine/SDK" as SDK
participant "Worker Process" as Worker
Client->>MCP: POST /mcp (JSON-RPC)
MCP->>MCP: Parse & dispatch by method
alt tools/list
MCP->>SDK: Query functions & metadata
SDK-->>MCP: Function list
MCP->>MCP: Filter by mcp.expose / expose_all
else tools/call (builtin)
MCP->>Worker: create/stop worker or trigger action
Worker-->>MCP: result/status
else tools/call (user)
MCP->>SDK: Trigger function call
SDK-->>MCP: function result
end
MCP-->>Client: JSON-RPC response
sequenceDiagram
participant Client
participant "A2A Handler" as A2A
participant "Task Store (state API)" as Store
participant "Engine/SDK" as SDK
Client->>A2A: POST /a2a (A2A JSON-RPC)
A2A->>A2A: Parse A2ARequest & dispatch
alt message/send
A2A->>Store: create Task (Working)
A2A->>SDK: resolve target function & trigger
SDK-->>A2A: function result / error
A2A->>Store: update Task (Completed/Failed + artifacts/history)
else tasks/get / tasks/list
A2A->>Store: load or list tasks
Store-->>A2A: task(s)
else tasks/cancel
A2A->>Store: load task, validate state
A2A->>Store: set Canceled and persist
end
A2A-->>Client: A2AResponse (result or error)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
rohitg00
force-pushed
the
feat/mcp-a2a-workers
branch
from
March 28, 2026 10:42
a9a9897 to
4230bb1
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
♻️ Duplicate comments (1)
connect/src/a2a/handler.rs (1)
127-131:⚠️ Potential issue | 🟠 MajorAgent card still advertises a hardcoded localhost endpoint.
This ignores the real deployment host/port and doesn't include
/a2a, so remote clients will discover an unusable interface.💡 Suggested fix
-async fn build_agent_card(iii: &III) -> AgentCard { +async fn build_agent_card(iii: &III, public_base_url: &str) -> AgentCard { AgentCard { // ... supported_interfaces: vec![AgentInterface { - url: "http://localhost:3111".to_string(), + url: format!("{}/a2a", public_base_url.trim_end_matches('/')), protocol_binding: "JSONRPC".to_string(), protocol_version: "0.3".to_string(), }],Thread
public_base_urlthroughA2AHandler::register(...)and theiii_a2abinary, rather than deriving it from the worker's control-plane URL.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/src/a2a/handler.rs` around lines 127 - 131, The Agent card currently hardcodes the interface URL to "http://localhost:3111" in the supported_interfaces AgentInterface, which should instead use the real public base URL and include the "/a2a" path; change A2AHandler::register to accept a public_base_url parameter and construct the AgentInterface.url as format!("{}/a2a", public_base_url.trim_end_matches('/')), update any usage-sites (notably the iii_a2a binary) to pass the actual public_base_url through to A2AHandler::register, and remove the localhost literal so remote clients discover the correct endpoint.
🧹 Nitpick comments (3)
connect/src/worker_manager/mod.rs (2)
123-130: Consider waiting for process exit before removing temp directory.The code kills the process then immediately removes the temp directory. If the kill is asynchronous or the process doesn't exit immediately,
remove_dir_allmay fail or leave orphaned files. Consider usingchild.wait()afterkill()to ensure the process has exited.♻️ Proposed fix
if let Err(e) = child.kill().await { tracing::warn!(worker_id = %params.id, error = %e, "Failed to kill worker process"); } + // Wait for process to actually exit before cleaning up + let _ = child.wait().await; if let Err(e) = tokio::fs::remove_dir_all(&info.temp_dir).await {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/src/worker_manager/mod.rs` around lines 123 - 130, After calling child.kill().await in the branch that removed the worker (the block using workers.remove(¶ms.id) and variable child), await the child process exit before removing the temp dir: call child.wait().await (or an appropriate wait_with_timeout) and handle/log any error from wait, then call tokio::fs::remove_dir_all(&info.temp_dir).await only after wait completes; keep the existing tracing::warn! calls to report kill/wait/remove errors and ensure you reference the same worker_id = %params.id and use the info.temp_dir and child variables when logging.
222-229: Consider capturing stderr for debugging worker startup failures.With both stdout and stderr piped to
Stdio::null(), diagnosing worker startup failures (import errors, syntax errors, SDK connection issues) will be very difficult. Consider capturing stderr and logging it, or providing a debug mode.♻️ Proposed approach
+ let stderr = if std::env::var("III_WORKER_DEBUG").is_ok() { + Stdio::inherit() + } else { + Stdio::null() + }; + Command::new(cmd) .arg(temp_dir.join(file_name)) .current_dir(temp_dir) .stdout(Stdio::null()) - .stderr(Stdio::null()) + .stderr(stderr) .kill_on_drop(true) .spawn()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/src/worker_manager/mod.rs` around lines 222 - 229, The worker startup currently sinks both stdout and stderr via Command::new(...).stdout(Stdio::null()).stderr(Stdio::null()) before .spawn(), which hides diagnostics; change stderr to be captured (e.g. .stderr(Stdio::piped()) or conditional on a debug flag), then after .spawn() read the child.stderr (or attach an async task) and surface its contents in the error path returned from spawn() or log it when the child exits non‑zero; make this behavior configurable (e.g. a debug or verbose flag passed into the worker manager) so you only capture/log stderr when helpful.connect/src/mcp/handler.rs (1)
310-327: Hardcodedsdk_versionwill become stale.Line 312 hardcodes
"sdk_version": "0.10.0"while line 390 correctly usesenv!("CARGO_PKG_VERSION"). Consider using the same approach for consistency.♻️ Proposed fix
"iii://context" => ( serde_json::to_string_pretty(&json!({ - "sdk_version": "0.10.0", + "sdk_version": env!("CARGO_PKG_VERSION"), "function_id_delimiter": "::",🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/src/mcp/handler.rs` around lines 310 - 327, The "iii://context" capability JSON hardcodes "sdk_version": "0.10.0" making it stale; replace the literal with the crate version macro used elsewhere (env!("CARGO_PKG_VERSION")) inside the serde_json! block so the value is compiled from CARGO_PKG_VERSION (same approach as used at the other match arm around line ~390); update the "iii://context" arm in handler.rs (the serde_json::to_string_pretty call building capabilities/metadata_filtering) to use env!("CARGO_PKG_VERSION") for sdk_version.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connect/src/a2a/handler.rs`:
- Around line 301-310: The code currently forwards any caller-supplied
function_id from resolve_function to iii.trigger, allowing hidden/internal
functions to be invoked; update the invocation path to apply the same exposure
predicate used in build_agent_card: after resolve_function(...) and before
calling iii.trigger(...), look up the function metadata (or function object) and
require EXPOSE_ALL || has_metadata_flag(function, "a2a.expose"); if the
predicate fails return an appropriate error/permission denied response instead
of calling iii.trigger. Ensure you reference the same helper/flag names
(EXPOSE_ALL, has_metadata_flag, build_agent_card, resolve_function, iii.trigger)
so the gate is identical to the one used during discovery.
- Around line 181-191: load_task currently swallows trigger and deserialization
errors and returns None, causing handle_get and handle_cancel to treat backend
failures or corrupted data as "task not found"; change load_task (symbol:
load_task) to return a Result<Option<Task>, E> (or Result<Task, E> where
appropriate) instead of Option<Task>, propagate or return trigger invocation
errors from iii.trigger (TriggerRequest, TASK_SCOPE) and deserialization errors
from serde_json::from_value, and update callers (handle_get, handle_cancel) to
match the new signature and map true NotFound vs internal errors to the correct
HTTP/error responses rather than conflating them.
- Around line 449-452: The current fallback returns ("state::get", json!{...})
which masks malformed input as an internal function call; instead, have the
parser/validator return an Err with a clear error (e.g. Err("Provide
function_id...")) where you currently produce the ("state::get", ...) tuple, and
update handle_send to convert that Err into an A2AResponse::error(...) (using
A2AResponse::error with the same message) so unsupported messages fail fast
rather than being passed to state::get.
- Around line 65-67: The serde_json::from_value failure for A2ARequest is
currently mapped to iii_sdk::IIIError::Runtime which bypasses the A2A/JSON-RPC
error envelope; update the map_err closure around serde_json::from_value(body)
so that parsing errors are converted into the A2A-specific error type/response
used by the project (e.g., construct and return the crate's A2A error/envelope
variant instead of IIIError::Runtime), using the existing A2A error
constructor/helper so the malformed request is sent back via the JSON-RPC/A2A
response path; locate the conversion near the A2ARequest deserialization line
and replace the map_err to produce the A2A error envelope rather than a runtime
error.
In `@connect/src/mcp/handler.rs`:
- Around line 543-548: The CallParams struct's #[serde(default)] makes arguments
become Value::Null when missing, so the generic function trigger that directly
passes params.arguments may send null instead of an empty object; fix this by
giving arguments a custom serde default that returns an empty JSON object (e.g.,
add #[serde(default = "default_arguments")] to CallParams::arguments and
implement a default_arguments() -> Value that returns json!({})), or
alternatively ensure the caller that forwards params.arguments (the generic
function trigger) replaces null with an empty object (e.g., use
params.arguments.or(json!({}))). Ensure you update references to CallParams and
the site that forwards params.arguments accordingly.
In `@connect/src/worker_manager/mod.rs`:
- Around line 180-190: The generated Python snippet calls
iii.register_function('{}', handler, '{}') which passes the description as a
positional argument and will raise a TypeError; update the format string so the
call uses the keyword argument description (iii.register_function('{}', handler,
description='{}')) by modifying the format template that emits the
register_function call (look for the format block that writes
iii.register_function and the variables '{}' and handler) to place description
as a keyword, ensuring the produced code matches the SDK signature.
---
Duplicate comments:
In `@connect/src/a2a/handler.rs`:
- Around line 127-131: The Agent card currently hardcodes the interface URL to
"http://localhost:3111" in the supported_interfaces AgentInterface, which should
instead use the real public base URL and include the "/a2a" path; change
A2AHandler::register to accept a public_base_url parameter and construct the
AgentInterface.url as format!("{}/a2a", public_base_url.trim_end_matches('/')),
update any usage-sites (notably the iii_a2a binary) to pass the actual
public_base_url through to A2AHandler::register, and remove the localhost
literal so remote clients discover the correct endpoint.
---
Nitpick comments:
In `@connect/src/mcp/handler.rs`:
- Around line 310-327: The "iii://context" capability JSON hardcodes
"sdk_version": "0.10.0" making it stale; replace the literal with the crate
version macro used elsewhere (env!("CARGO_PKG_VERSION")) inside the serde_json!
block so the value is compiled from CARGO_PKG_VERSION (same approach as used at
the other match arm around line ~390); update the "iii://context" arm in
handler.rs (the serde_json::to_string_pretty call building
capabilities/metadata_filtering) to use env!("CARGO_PKG_VERSION") for
sdk_version.
In `@connect/src/worker_manager/mod.rs`:
- Around line 123-130: After calling child.kill().await in the branch that
removed the worker (the block using workers.remove(¶ms.id) and variable
child), await the child process exit before removing the temp dir: call
child.wait().await (or an appropriate wait_with_timeout) and handle/log any
error from wait, then call tokio::fs::remove_dir_all(&info.temp_dir).await only
after wait completes; keep the existing tracing::warn! calls to report
kill/wait/remove errors and ensure you reference the same worker_id = %params.id
and use the info.temp_dir and child variables when logging.
- Around line 222-229: The worker startup currently sinks both stdout and stderr
via Command::new(...).stdout(Stdio::null()).stderr(Stdio::null()) before
.spawn(), which hides diagnostics; change stderr to be captured (e.g.
.stderr(Stdio::piped()) or conditional on a debug flag), then after .spawn()
read the child.stderr (or attach an async task) and surface its contents in the
error path returned from spawn() or log it when the child exits non‑zero; make
this behavior configurable (e.g. a debug or verbose flag passed into the worker
manager) so you only capture/log stderr when helpful.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 4546a3d6-b80e-4c73-bbe6-b4e27155ceb2
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (15)
Cargo.tomlconnect/Cargo.tomlconnect/src/a2a/handler.rsconnect/src/a2a/mod.rsconnect/src/a2a/types.rsconnect/src/bin/iii_a2a.rsconnect/src/bin/iii_mcp.rsconnect/src/json_rpc.rsconnect/src/lib.rsconnect/src/mcp/handler.rsconnect/src/mcp/mod.rsconnect/src/mcp/prompts.rsconnect/src/transport/mod.rsconnect/src/transport/stdio.rsconnect/src/worker_manager/mod.rs
✅ Files skipped from review due to trivial changes (7)
- Cargo.toml
- connect/src/mcp/mod.rs
- connect/src/transport/mod.rs
- connect/src/a2a/mod.rs
- connect/src/lib.rs
- connect/Cargo.toml
- connect/src/json_rpc.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- connect/src/mcp/prompts.rs
- connect/src/transport/stdio.rs
- connect/src/a2a/types.rs
Comment on lines
+181
to
+191
| async fn load_task(iii: &III, task_id: &str) -> Option<Task> { | ||
| iii.trigger(TriggerRequest { | ||
| function_id: "state::get".to_string(), | ||
| payload: json!({ "scope": TASK_SCOPE, "key": task_id }), | ||
| action: None, | ||
| timeout_ms: Some(5000), | ||
| }) | ||
| .await | ||
| .ok() | ||
| .and_then(|v| serde_json::from_value(v).ok()) | ||
| } |
Contributor
There was a problem hiding this comment.
Don't collapse state-store failures into Task not found.
load_task maps trigger errors and decode errors to None, so handle_get and handle_cancel will misreport backend outages or corrupt stored data as a missing task.
💡 Suggested fix
-async fn load_task(iii: &III, task_id: &str) -> Option<Task> {
- iii.trigger(TriggerRequest {
- function_id: "state::get".to_string(),
- payload: json!({ "scope": TASK_SCOPE, "key": task_id }),
- action: None,
- timeout_ms: Some(5000),
- })
- .await
- .ok()
- .and_then(|v| serde_json::from_value(v).ok())
+async fn load_task(iii: &III, task_id: &str) -> Result<Option<Task>, iii_sdk::IIIError> {
+ let value = iii
+ .trigger(TriggerRequest {
+ function_id: "state::get".to_string(),
+ payload: json!({ "scope": TASK_SCOPE, "key": task_id }),
+ action: None,
+ timeout_ms: Some(5000),
+ })
+ .await?;
+
+ if value.is_null() {
+ return Ok(None);
+ }
+
+ let task = serde_json::from_value(value)
+ .map_err(|e| iii_sdk::IIIError::Runtime(format!("Invalid stored task: {e}")))?;
+ Ok(Some(task))
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/src/a2a/handler.rs` around lines 181 - 191, load_task currently
swallows trigger and deserialization errors and returns None, causing handle_get
and handle_cancel to treat backend failures or corrupted data as "task not
found"; change load_task (symbol: load_task) to return a Result<Option<Task>, E>
(or Result<Task, E> where appropriate) instead of Option<Task>, propagate or
return trigger invocation errors from iii.trigger (TriggerRequest, TASK_SCOPE)
and deserialization errors from serde_json::from_value, and update callers
(handle_get, handle_cancel) to match the new signature and map true NotFound vs
internal errors to the correct HTTP/error responses rather than conflating them.
Contributor
Author
There was a problem hiding this comment.
Fair point but the engine's state::get returns null for missing keys and errors for actual failures. serde_json::from_value(null) fails deserialization which maps to None. In practice, storage failures are distinguishable from missing tasks at the engine level. Adding a separate error path here would require changing the return type for a case that doesn't happen in normal operation.
Contributor
There was a problem hiding this comment.
@rohitg00, that's a helpful clarification on the engine contract — state::get returning null for missing keys vs. Err for actual failures is a well-defined distinction at the engine level.
The one remaining gap is that at the Rust call-site, .ok() on the trigger future still converts engine-level Errs into None before they reach handle_get/handle_cancel, so a transient backend outage would still surface to the caller as "task not found" rather than an internal error. That said, I understand you're accepting this tradeoff intentionally given the rarity of real storage failures in practice, consistent with the broader fire-and-forget approach in this module.
Noted — won't flag this further.
✏️ Learnings added
Learnt from: rohitg00
Repo: iii-hq/iii PR: 1374
File: connect/a2a/src/handler.rs:190-200
Timestamp: 2026-03-28T11:20:59.196Z
Learning: In `connect/a2a/src/handler.rs`, `load_task` intentionally uses `.ok().and_then(...)` to collapse both missing-key (engine returns JSON null → deserialization fails → None) and trigger errors (rare storage failures) into `Option<Task>`. The engine's `state::get` contract is: returns JSON null for a missing key, returns Err for actual failures. The author deliberately accepts that trigger errors are silently mapped to None (task not found) for simplicity, as real storage failures are considered non-existent in normal operation. Do not flag this as a bug.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: rohitg00
Repo: iii-hq/iii PR: 1374
File: connect/a2a/src/handler.rs:176-188
Timestamp: 2026-03-28T11:17:45.432Z
Learning: In iii-hq/iii, `store_task` in `connect/a2a/src/handler.rs` intentionally uses `TriggerAction::Void` (fire-and-forget) for performance. A2A task state is transient, not durable — acknowledged writes are deliberately avoided to reduce per-request latency. Errors are logged but not surfaced to callers. Do not flag this as a bug.
Learnt from: rohitg00
Repo: iii-hq/iii PR: 1374
File: connect/a2a/src/handler.rs:410-426
Timestamp: 2026-03-28T11:17:52.303Z
Learning: In `connect/a2a/src/handler.rs` (Rust), `handle_cancel` has a known TOCTOU race between `load_task` and `store_task`: a concurrent `handle_send` can complete the task between the state check and the store, causing a completed task to be marked as canceled. This is intentionally documented as eventual consistency behavior for now. A proper atomic compare-and-swap fix is deferred pending engine-level CAS state primitives being introduced in PRs `#1371/`#1373 (RBAC work).
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/modules/module-stream.mdx:349-351
Timestamp: 2026-03-16T11:50:55.021Z
Learning: In iii-hq/iii, `RegisterTriggerInput` (sdk/packages/rust/iii/src/protocol.rs:208) has exactly three required fields — `trigger_type: String`, `function_id: String`, `config: Value` — and NO optional fields. It does NOT implement Default. All Rust doc examples must use `RegisterTriggerInput { trigger_type: "...".into(), function_id: "...".into(), config: json!({...}) }` without `..Default::default()`. The docs incorrectly used `type_:` instead of `trigger_type:` in files modified by PR `#1313` (module-stream.mdx, architecture/trigger-types.mdx, advanced/sdk-implementation.mdx, examples/conditions.mdx, examples/cron.mdx, examples/hello-world.mdx, examples/observability.mdx).
Learnt from: anthonyiscoding
Repo: iii-hq/iii PR: 1331
File: website/components/sections/code-examples/iii/jobs.ts:32-38
Timestamp: 2026-03-19T20:02:17.308Z
Learning: In iii-hq/iii (website/components/sections/code-examples/iii/), not every `iii.trigger()` call needs to be awaited. Some trigger invocations (e.g., with `TriggerAction.Enqueue`) are intentionally fire-and-forget. Do not flag unawaited `iii.trigger()` calls as bugs — the author may deliberately omit `await` for async/non-blocking dispatch patterns.
Comment on lines
+180
to
+190
| format!( | ||
| r#"import asyncio | ||
| import signal | ||
| from iii_sdk import register_worker, Logger | ||
|
|
||
| iii = register_worker('{}') | ||
| logger = Logger() | ||
|
|
||
| {} | ||
|
|
||
| iii.register_function('{}', handler, '{}') |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check Python SDK exports and register_function signature
fd -t f '__init__.py' --full-path '**/sdk/packages/python/**' --exec cat {}Repository: iii-hq/iii
Length of output: 166
🏁 Script executed:
# Find Python SDK related files
git ls-files | grep -E 'sdk.*python' | head -20Repository: iii-hq/iii
Length of output: 956
🏁 Script executed:
# Search for register_worker and register_function definitions
rg -l 'register_worker|register_function' --type pyRepository: iii-hq/iii
Length of output: 1949
🏁 Script executed:
# Look for SDK package structure
fd -type d -name 'sdk' -o -name 'python' | head -20Repository: iii-hq/iii
Length of output: 224
🏁 Script executed:
# Check Python SDK main module exports
cat -n sdk/packages/python/iii/src/iii/__init__.py | head -100Repository: iii-hq/iii
Length of output: 3230
🏁 Script executed:
# Check the iii.py implementation for register_worker and register_function
cat -n sdk/packages/python/iii/src/iii/iii.py | head -150Repository: iii-hq/iii
Length of output: 6941
🏁 Script executed:
# Check test file for register_function arguments
cat -n sdk/packages/python/iii/tests/test_register_function_args.pyRepository: iii-hq/iii
Length of output: 17156
🏁 Script executed:
# Find register_function method definition in iii.py
rg -A 20 "def register_function" sdk/packages/python/iii/src/iii/iii.py | head -100Repository: iii-hq/iii
Length of output: 1046
🏁 Script executed:
# Find register_worker function definition
rg -A 15 "def register_worker" sdk/packages/python/iii/src/iii/iii.pyRepository: iii-hq/iii
Length of output: 608
Generated Python code has incorrect register_function() call signature.
The generated code calls iii.register_function('{}', handler, '{}'), passing description as a positional argument. However, the Python SDK's register_function() method signature only accepts two positional arguments (function ID and handler/invocation config) before its keyword-only separator (*). Description must be passed as a keyword argument.
The correct call should be:
iii.register_function('{}', handler, description='{}')Without this fix, the generated code will raise TypeError at runtime.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/src/worker_manager/mod.rs` around lines 180 - 190, The generated
Python snippet calls iii.register_function('{}', handler, '{}') which passes the
description as a positional argument and will raise a TypeError; update the
format string so the call uses the keyword argument description
(iii.register_function('{}', handler, description='{}')) by modifying the format
template that emits the register_function call (look for the format block that
writes iii.register_function and the variables '{}' and handler) to place
description as a keyword, ensuring the produced code matches the SDK signature.
Contributor
Author
There was a problem hiding this comment.
Fixed in c60efcf — same escaping applied to the Python template. Both function_name and description are escaped before interpolation.
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
git show c60efcf5 -- connect/mcp/src/worker_manager.rs 2>/dev/null || git show c60efcf5 2>/dev/null | head -100
Length of output: 2009
🏁 Script executed:
#!/bin/bash
rg -n "register_function" connect/mcp/src/worker_manager.rs -A 2 -B 2
Length of output: 93
@rohitg00, the escaping for function_name and description looks good in commit c60efcf. However, the original issue — passing description as a keyword argument — is still not fixed. Line 202 of the current file still reads:
iii.register_function('{}', handler, '{}')The description is still the third positional argument, which will raise a TypeError at runtime since the Python SDK's register_function only accepts two positional arguments before the keyword-only separator. It needs to be:
iii.register_function('{}', handler, description='{}')The format template in generate_python_worker needs this update:
format!(
r#"...
iii.register_function('{}', handler, description='{}')
..."#,
self.engine_url, params.code, function_name, description, function_name
)🧠 Learnings used
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/modules/module-stream.mdx:349-351
Timestamp: 2026-03-16T11:50:55.021Z
Learning: In iii-hq/iii Rust SDK docs, `RegisterFunctionMessage` can be constructed using `RegisterFunctionMessage::new("id")` (a constructor method added to `sdk/packages/rust/iii/src/protocol.rs` in PR `#1313` follow-up). Use this pattern in all doc examples where only the `id` field is needed. When `description` or other optional fields must be set, use the full struct literal with explicit `None` for unset fields (e.g., `request_format: None, response_format: None, metadata: None, invocation: None`). Never use `..Default::default()` as `RegisterFunctionMessage` does not implement `Default`.
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/modules/module-stream.mdx:349-351
Timestamp: 2026-03-16T11:50:55.021Z
Learning: In iii-hq/iii, `RegisterFunctionMessage::new(id)` was added as a constructor method in `sdk/packages/rust/iii/src/protocol.rs` as part of the fix for PR `#1313`. All Rust doc examples that only set `id` should use `RegisterFunctionMessage::new("X")`. When other optional fields (description, request_format, response_format, metadata, invocation) must be set, use the full struct literal with explicit `None` for unset fields. Never use `..Default::default()` since `RegisterFunctionMessage` does not implement `Default`.
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/modules/module-stream.mdx:349-351
Timestamp: 2026-03-16T11:50:55.021Z
Learning: In iii-hq/iii, `RegisterTriggerInput` (sdk/packages/rust/iii/src/protocol.rs:208) has exactly three required fields — `trigger_type: String`, `function_id: String`, `config: Value` — and NO optional fields. It does NOT implement Default. All Rust doc examples must use `RegisterTriggerInput { trigger_type: "...".into(), function_id: "...".into(), config: json!({...}) }` without `..Default::default()`. The docs incorrectly used `type_:` instead of `trigger_type:` in files modified by PR `#1313` (module-stream.mdx, architecture/trigger-types.mdx, advanced/sdk-implementation.mdx, examples/conditions.mdx, examples/cron.mdx, examples/hello-world.mdx, examples/observability.mdx).
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/modules/module-stream.mdx:349-351
Timestamp: 2026-03-16T11:50:55.021Z
Learning: Guideline for iii-hq/iii Rust SDK docs: `RegisterFunctionMessage` and `RegisterTriggerInput` do NOT implement `Default` (they only derive `Debug, Clone, Serialize, Deserialize`). Never use `..Default::default()` when constructing these structs in any documentation example under `docs/content/`. Instead, always explicitly set every optional field to `None` (e.g., `description: None, request_format: None, response_format: None, metadata: None, invocation: None` for `RegisterFunctionMessage`). This applies to all `.mdx` files across `docs/content/examples/`, `docs/content/architecture/`, `docs/content/advanced/`, and `docs/content/modules/`.
Learnt from: guibeira
Repo: iii-hq/iii PR: 1313
File: docs/content/how-to/create-ephemeral-worker.mdx:172-172
Timestamp: 2026-03-16T11:41:45.946Z
Learning: In the iii SDK (Rust), `shutdown()` is deprecated in favor of `shutdown_async().await`. The reason is to guarantee telemetry flush before the process exits. Documentation examples and prose in docs/content/how-to/create-ephemeral-worker.mdx and other Rust SDK docs should use `shutdown_async().await` instead of `shutdown()`.
Two standalone iii workers as separate crates under connect/.
No shared library. Each has its own Cargo.toml and binary.
connect/mcp/ — iii-mcp v0.3.0
- MCP spec 2025-11-25
- stdio transport for Claude Desktop / Cursor
- Streamable HTTP at POST /mcp on engine port
- tools, resources, prompts, notifications/tools/list_changed
- 6 builtin tools: worker register/stop, trigger register/unregister, void, enqueue
- metadata filtering: functions need { "mcp.expose": true } in metadata
- --expose-all flag to show all functions
connect/a2a/ — iii-a2a v0.3.0
- A2A v0.3 (v1.0 method names accepted)
- Agent Card at GET /.well-known/agent-card.json
- JSON-RPC at POST /a2a
- message/send, tasks/get, tasks/cancel, tasks/list
- Task state stored in engine KV, ISO 8601 timestamps
- metadata filtering: functions need { "a2a.expose": true } in metadata
- --expose-all flag to show all functions
rohitg00
force-pushed
the
feat/mcp-a2a-workers
branch
from
March 28, 2026 10:59
4230bb1 to
b6d1738
Compare
…emplate params, A2A error envelope
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
♻️ Duplicate comments (11)
connect/a2a/src/handler.rs (6)
451-454:⚠️ Potential issue | 🟠 MajorUnsupported messages should return an A2A error.
Falling back to
state::getturns malformed input into a synthetic internal call, so the task may come backcompletedwith meaningless output instead of failing validation. Return an error fromresolve_function()and map that toA2AResponse::error(...)inhandle_send().🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 451 - 454, The current resolve_function() returns a synthetic tuple ("state::get", json!({...})) for unsupported messages which causes malformed input to be treated as an internal call; instead modify resolve_function() to return an Err describing the validation failure (e.g., missing function_id) rather than the ("state::get", ...) tuple, and update handle_send() to map that Err into an A2AResponse::error(...) (using the error message and appropriate scope/key) so unsupported messages produce a proper A2A error response rather than a synthetic state::get call.
129-133:⚠️ Potential issue | 🟠 MajorAdvertise the actual A2A endpoint URL.
AgentInterface.urlis hardcoded tohttp://localhost:3111, whileconnect/a2a/src/main.rsalready parses--engine-urlon Lines 13-14. Any remote host or non-default port will get a bad agent card unless this is derived from the configured address instead of localhost.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 129 - 133, The AgentInterface.url is hardcoded to "http://localhost:3111" in the supported_interfaces declaration; change it to use the configured engine URL parsed in main (the --engine-url value) so the agent card advertises the real A2A endpoint. Locate the supported_interfaces vec and replace the literal with the runtime/config value (the parsed engine_url or equivalent) so AgentInterface.url is constructed from that variable (preserving protocol and port), ensuring AgentInterface and supported_interfaces reflect the configured address rather than localhost.
303-312:⚠️ Potential issue | 🔴 CriticalEnforce
a2a.exposebefore invoking the function.
build_agent_card()hides non-exposed functions, but this path still lets the caller choose anyfunction_idand forwards it toiii.trigger(...). Internal functions remain callable over/a2aunless you apply the same exposure predicate here.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 303 - 312, The handler currently calls iii.trigger with any function_id resolved by resolve_function, bypassing the exposure check used by build_agent_card; before calling iii.trigger (and after resolve_function), validate that the resolved function_id is exposed via the same a2a.expose predicate used in build_agent_card and return an appropriate error/HTTP 403 if not exposed; reference the resolve_function result (function_id / fn_name) and the iii.trigger call to locate the insertion point and reuse the exposure predicate logic to enforce the restriction.
403-419:⚠️ Potential issue | 🟡 MinorCancel is racy against concurrent completion.
This load-check-store sequence can overwrite a task that completed between the
load_task()call and the finalstore_task(). Use a compare-and-swap update in the state layer, or document this as eventual consistency if that is the intended contract.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 403 - 419, The current cancel flow loads a Task via load_task(...) then unconditionally writes a Canceled TaskStatus with store_task(...), which can overwrite a terminal Completed state that arrived concurrently; replace the load-check-store with an atomic compare-and-swap update in the persistence layer (e.g., add/update and use an update_task_if_state/update_task_cas API) so the write only succeeds if the task's state still matches the expected non-terminal state, and return A2AResponse::error(id, -32002, ...) if the CAS fails because the state changed; alternatively, perform the update inside a transaction in the store_task implementation and surface a failure to the handler rather than blindly storing TaskStatus from the stale load_task result.
183-193:⚠️ Potential issue | 🟠 MajorDon’t collapse storage errors into “not found.”
.ok().and_then(...)turns trigger failures and bad stored JSON intoNone. Downstream,tasks/getandtasks/cancelthen reportTask not foundfor backend outages or corrupt state.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 183 - 193, The load_task helper collapses trigger failures and JSON parse errors into None, causing downstream handlers (tasks/get, tasks/cancel) to report "Task not found" for infrastructure or corrupt-state errors; change load_task (and its callers) to return a Result<Option<Task>, E> (e.g., anyhow::Result<Option<Task>>) instead of Option<Task>, propagate and return Err when iii.trigger(...) returns Err or when serde_json::from_value(...) fails, and leave Ok(None) only for a genuine missing value, so callers can distinguish NotFound vs backend/parse errors (refer to function load_task, the iii.trigger call and serde_json::from_value usage, and update tasks/get and tasks/cancel to handle Err vs Ok(None) accordingly).
169-180:⚠️ Potential issue | 🟠 MajorPersist tasks with an acknowledged write.
TriggerAction::Voidonly confirms that the invoke frame was sent; it does not confirm thatstate::setsucceeded.message/sendandtasks/cancelcan therefore return a task that the state store never actually committed.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 169 - 180, store_task currently invokes the state::set trigger with TriggerAction::Void which only confirms the invoke frame was sent and not that the state write committed; change the TriggerRequest to use TriggerAction::Ack (or the SDK's equivalent acknowledgement action) so the call waits for the state::set operation to be acknowledged, keep the existing error handling on the resulting await, and consider adding a sensible timeout_ms value on the TriggerRequest to avoid hanging; update references in store_task to construct TriggerRequest { function_id: "state::set", payload: ..., action: TriggerAction::Ack, timeout_ms: Some(...) } so callers like message/send and tasks/cancel get a truly persisted task.connect/mcp/src/worker_manager.rs (2)
180-190:⚠️ Potential issue | 🔴 CriticalUse
description=in the generated Pythonregister_function()call.Line 190 passes the description as a third positional argument. The Python SDK expects a keyword here, so generated Python workers will fail before registration.
Suggested fix
-iii.register_function('{}', handler, '{}') +iii.register_function('{}', handler, description='{}')Verify against the current SDK definition with:
#!/bin/bash set -euo pipefail fd -i '^iii\.py$' sdk/packages/python -x rg -n -C5 'def register_function' {} sed -n '180,191p' connect/mcp/src/worker_manager.rsExpected result: the SDK signature shows
descriptionis keyword-only, while the template currently emits a third positional argument.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 180 - 190, The Python template in connect/mcp/src/worker_manager.rs currently emits iii.register_function('{}', handler, '{}') which passes the description as a positional argument; update the format string to call iii.register_function('{}', handler, description='{}') so the description is passed as a keyword-only argument (ensure the same format placeholders are preserved and any surrounding format! call still interpolates the worker name and description correctly).
143-170:⚠️ Potential issue | 🔴 Critical
iii_worker_registeris still a host-level code-execution primitive.
params.codeis executed verbatim, andfunction_name/descriptionare interpolated into both templates without escaping. If this MCP surface is reachable by anything except a fully trusted admin, a caller can break the generated source or run arbitrary code on the worker host.Also applies to: 174-206
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 143 - 170, generate_node_worker currently interpolates params.code, params.function_name and params.description directly into a JS template, allowing code injection; update generate_node_worker to treat untrusted inputs safely by (a) not executing params.code verbatim — embed it as a safely quoted/serialized string or store it externally and reference it, and (b) escape/serialize params.function_name and params.description (e.g., JSON/string literal escaping) before interpolation so they cannot break the generated source; ensure the same fixes are applied to the other Node template generation block that uses params.code/params.function_name/params.description.connect/mcp/src/handler.rs (2)
415-460:⚠️ Potential issue | 🟠 MajorHTTP mode still advertises capabilities it cannot fulfill.
dispatch_http()includesbuiltin_tools()intools/list, buttools/callthere never handles those builtins; it only forwards toiii.trigger(...). This path also ignoresexpose_alland omitsresources/read/resources/templates/list, so--no-stdioexposes a materially different MCP surface.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/handler.rs` around lines 415 - 460, The HTTP handler advertises built-in tools and extra resources it doesn't actually support: update the "tools/list" and "tools/call" handling so they are consistent. Either remove builtin_tools() from the "tools/list" branch or implement builtin handling in the "tools/call" branch (recognize the same tool IDs and execute the built-in behavior instead of only calling iii.trigger), ensure listing honors the expose_all flag and has_metadata_flag(...) filtering used elsewhere (so "tools/list" only advertises what will be handled), and add handlers for "resources/read" and "resources/templates/list" (or stop advertising them) so the set of methods returned by builtin_tools()/resources list matches the methods implemented in the match (check functions: builtin_tools(), has_metadata_flag, the "tools/call" branch that calls iii.trigger(TriggerRequest { ... }), and the resources-related method names).
244-245:⚠️ Potential issue | 🟡 MinorMissing
argumentsstill deserializes tonull.
#[serde(default)]onserde_json::ValueyieldsValue::Null, and both generic trigger paths forward that value unchanged. Functions expecting an object will receivenullwhenever the client omitsarguments.Suggested fix
#[derive(Deserialize)] struct CallParams { name: String, - #[serde(default)] + #[serde(default = "default_arguments")] arguments: Value, } + +fn default_arguments() -> Value { + json!({}) +}Also applies to: 443-444, 531-535
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/handler.rs` around lines 244 - 245, The code currently uses serde_json::Value with #[serde(default)] so missing "arguments" becomes Value::Null; change the parameter field type from serde_json::Value to Option<serde_json::Value> in the request/params struct(s) and update usages (e.g. where you build payload: params.arguments) to handle None explicitly (either propagate Option or convert None to an empty object using serde_json::Map/new when a consumer expects an object). Apply the same change to the other occurrences noted (around the other locations ~443-444 and ~531-535) so callers no longer receive Value::Null for omitted arguments.connect/mcp/src/prompts.rs (1)
35-51:⚠️ Potential issue | 🟡 Minor
prompts/getstill contradicts the required-argument schema.
list()marks these fields as required, butget()silently invents defaults instead of rejecting missing values. That makes caller mistakes look successful and weakens the advertised prompt contract.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/prompts.rs` around lines 35 - 51, The get() prompt builder is silently inventing defaults for required fields (e.g., lang/fid in the "register-function" arm, method/path in "build-api", schedule in "setup-cron") instead of failing per the schema; update the prompts::get implementation to validate required args (check args.get("language"), args.get("function_id"), args.get("method"), args.get("path"), args.get("schedule") as appropriate) and return a clear error (or propagate a validation error) when any required field is missing rather than using unwrap_or defaults (remove the unwrap_or("...") usage for those required keys and replace with explicit presence checks and error handling).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connect/a2a/src/handler.rs`:
- Around line 20-21: Change register(iii: &III, expose_all: bool) to return
Result<(), E> (choose an appropriate error type or Box<dyn Error>) and propagate
errors from register_trigger(...) instead of only logging them; set
EXPOSE_ALL.store(...) as before, call register_trigger(...)? to bubble up
failures, and ensure the caller (main) handles the Err path so "A2A endpoints
registered" is only logged on Ok. Update any call sites (e.g., main) to handle
the Result and fail fast on errors.
In `@connect/a2a/src/main.rs`:
- Around line 43-50: register_worker(&args.engine_url, InitOptions::default())
starts background work and telemetry but the code returns immediately after
Ctrl+C; call iii.shutdown_async().await before returning to gracefully stop the
background connection and flush telemetry. Locate the variable iii (from
register_worker) and insert an awaited call to its shutdown_async() method just
prior to returning Ok(()) so shutdown_async().await runs after
tokio::signal::ctrl_c().await.
In `@connect/mcp/src/handler.rs`:
- Around line 239-247: tools/call currently converts params.name into
function_id and directly calls self.iii.trigger(TriggerRequest{...}) which
bypasses the mcp.expose gate; add an explicit exposure check using the same mcp
exposure/metadata check used elsewhere before constructing the TriggerRequest
(e.g., verify that function_id derived from params.name is exposed according to
the MCP metadata and return an error if not), then only call self.iii.trigger
when the check passes; apply the same fix to the other occurrence that maps
params.name -> function_id and calls self.iii.trigger (the block around lines
439-446) so hidden functions cannot be invoked when --expose-all is off.
- Around line 332-341: The handler is returning a locally-generated UUID instead
of the engine-side trigger ID, which breaks cross-session trigger management;
update the code that calls self.iii.register_trigger (and similarly the block at
357-362) to use the engine-provided ID: either modify the Trigger type
(sdk/packages/rust/iii/src/triggers.rs) to expose its engine-side id field or
change register_trigger to return a tuple/struct that contains both the engine
id and the handle, then store the trigger handle in self.triggers keyed by the
engine id (not a new uuid) and return that engine id in the tool_json response
so unregister/list operations work across sessions (ensure you update usages of
register_trigger, self.triggers.insert, and the tool_json payload accordingly).
In `@connect/mcp/src/transport.rs`:
- Around line 16-24: The current loop drains handler.take_notification() but
then blocks on reader.read_line(&mut line).await so subsequent
handler.take_notification() notifications (e.g.,
notifications/tools/list_changed) are delayed until the next stdin line; fix by
making the read and notification waits concurrent — either: (A) use
tokio::select! between a futures-aware notification future
(handler.take_notification()) and reader.read_line(&mut line) so you respond
immediately to whichever completes (on notification branch write via
writer.write_all/flush), or (B) move notification handling into a dedicated task
that owns writer and consumes notifications from handler, while the main task
only awaits read_line; update the loop to use the chosen approach and ensure
writer access is synchronized/owned by the notification task and that errors
from writer.write_all/flush are handled.
In `@connect/mcp/src/worker_manager.rs`:
- Around line 65-89: create_worker() currently creates temp_dir and writes files
before error paths, leaving orphaned artifacts on unsupported language, write
failures, or spawn_worker() errors; update create_worker() to
remove_dir_all(&temp_dir) on any early Err paths (unsupported language branch,
tokio::fs::write map_err, and any error returned from spawn_worker()) so the
temp dir is cleaned up before returning the error. Implement this by calling
tokio::fs::remove_dir_all(&temp_dir).await.unwrap_or_default() (or
propagate/remove errors appropriately) in each error branch or wrap the
creation+work in a scope that uses a guard/finally-style cleanup if the Result
is Err; reference temp_dir, file_path, params.language, create_worker(),
spawn_worker(), generate_node_worker(), and generate_python_worker() to locate
where to add the cleanup.
---
Duplicate comments:
In `@connect/a2a/src/handler.rs`:
- Around line 451-454: The current resolve_function() returns a synthetic tuple
("state::get", json!({...})) for unsupported messages which causes malformed
input to be treated as an internal call; instead modify resolve_function() to
return an Err describing the validation failure (e.g., missing function_id)
rather than the ("state::get", ...) tuple, and update handle_send() to map that
Err into an A2AResponse::error(...) (using the error message and appropriate
scope/key) so unsupported messages produce a proper A2A error response rather
than a synthetic state::get call.
- Around line 129-133: The AgentInterface.url is hardcoded to
"http://localhost:3111" in the supported_interfaces declaration; change it to
use the configured engine URL parsed in main (the --engine-url value) so the
agent card advertises the real A2A endpoint. Locate the supported_interfaces vec
and replace the literal with the runtime/config value (the parsed engine_url or
equivalent) so AgentInterface.url is constructed from that variable (preserving
protocol and port), ensuring AgentInterface and supported_interfaces reflect the
configured address rather than localhost.
- Around line 303-312: The handler currently calls iii.trigger with any
function_id resolved by resolve_function, bypassing the exposure check used by
build_agent_card; before calling iii.trigger (and after resolve_function),
validate that the resolved function_id is exposed via the same a2a.expose
predicate used in build_agent_card and return an appropriate error/HTTP 403 if
not exposed; reference the resolve_function result (function_id / fn_name) and
the iii.trigger call to locate the insertion point and reuse the exposure
predicate logic to enforce the restriction.
- Around line 403-419: The current cancel flow loads a Task via load_task(...)
then unconditionally writes a Canceled TaskStatus with store_task(...), which
can overwrite a terminal Completed state that arrived concurrently; replace the
load-check-store with an atomic compare-and-swap update in the persistence layer
(e.g., add/update and use an update_task_if_state/update_task_cas API) so the
write only succeeds if the task's state still matches the expected non-terminal
state, and return A2AResponse::error(id, -32002, ...) if the CAS fails because
the state changed; alternatively, perform the update inside a transaction in the
store_task implementation and surface a failure to the handler rather than
blindly storing TaskStatus from the stale load_task result.
- Around line 183-193: The load_task helper collapses trigger failures and JSON
parse errors into None, causing downstream handlers (tasks/get, tasks/cancel) to
report "Task not found" for infrastructure or corrupt-state errors; change
load_task (and its callers) to return a Result<Option<Task>, E> (e.g.,
anyhow::Result<Option<Task>>) instead of Option<Task>, propagate and return Err
when iii.trigger(...) returns Err or when serde_json::from_value(...) fails, and
leave Ok(None) only for a genuine missing value, so callers can distinguish
NotFound vs backend/parse errors (refer to function load_task, the iii.trigger
call and serde_json::from_value usage, and update tasks/get and tasks/cancel to
handle Err vs Ok(None) accordingly).
- Around line 169-180: store_task currently invokes the state::set trigger with
TriggerAction::Void which only confirms the invoke frame was sent and not that
the state write committed; change the TriggerRequest to use TriggerAction::Ack
(or the SDK's equivalent acknowledgement action) so the call waits for the
state::set operation to be acknowledged, keep the existing error handling on the
resulting await, and consider adding a sensible timeout_ms value on the
TriggerRequest to avoid hanging; update references in store_task to construct
TriggerRequest { function_id: "state::set", payload: ..., action:
TriggerAction::Ack, timeout_ms: Some(...) } so callers like message/send and
tasks/cancel get a truly persisted task.
In `@connect/mcp/src/handler.rs`:
- Around line 415-460: The HTTP handler advertises built-in tools and extra
resources it doesn't actually support: update the "tools/list" and "tools/call"
handling so they are consistent. Either remove builtin_tools() from the
"tools/list" branch or implement builtin handling in the "tools/call" branch
(recognize the same tool IDs and execute the built-in behavior instead of only
calling iii.trigger), ensure listing honors the expose_all flag and
has_metadata_flag(...) filtering used elsewhere (so "tools/list" only advertises
what will be handled), and add handlers for "resources/read" and
"resources/templates/list" (or stop advertising them) so the set of methods
returned by builtin_tools()/resources list matches the methods implemented in
the match (check functions: builtin_tools(), has_metadata_flag, the "tools/call"
branch that calls iii.trigger(TriggerRequest { ... }), and the resources-related
method names).
- Around line 244-245: The code currently uses serde_json::Value with
#[serde(default)] so missing "arguments" becomes Value::Null; change the
parameter field type from serde_json::Value to Option<serde_json::Value> in the
request/params struct(s) and update usages (e.g. where you build payload:
params.arguments) to handle None explicitly (either propagate Option or convert
None to an empty object using serde_json::Map/new when a consumer expects an
object). Apply the same change to the other occurrences noted (around the other
locations ~443-444 and ~531-535) so callers no longer receive Value::Null for
omitted arguments.
In `@connect/mcp/src/prompts.rs`:
- Around line 35-51: The get() prompt builder is silently inventing defaults for
required fields (e.g., lang/fid in the "register-function" arm, method/path in
"build-api", schedule in "setup-cron") instead of failing per the schema; update
the prompts::get implementation to validate required args (check
args.get("language"), args.get("function_id"), args.get("method"),
args.get("path"), args.get("schedule") as appropriate) and return a clear error
(or propagate a validation error) when any required field is missing rather than
using unwrap_or defaults (remove the unwrap_or("...") usage for those required
keys and replace with explicit presence checks and error handling).
In `@connect/mcp/src/worker_manager.rs`:
- Around line 180-190: The Python template in connect/mcp/src/worker_manager.rs
currently emits iii.register_function('{}', handler, '{}') which passes the
description as a positional argument; update the format string to call
iii.register_function('{}', handler, description='{}') so the description is
passed as a keyword-only argument (ensure the same format placeholders are
preserved and any surrounding format! call still interpolates the worker name
and description correctly).
- Around line 143-170: generate_node_worker currently interpolates params.code,
params.function_name and params.description directly into a JS template,
allowing code injection; update generate_node_worker to treat untrusted inputs
safely by (a) not executing params.code verbatim — embed it as a safely
quoted/serialized string or store it externally and reference it, and (b)
escape/serialize params.function_name and params.description (e.g., JSON/string
literal escaping) before interpolation so they cannot break the generated
source; ensure the same fixes are applied to the other Node template generation
block that uses params.code/params.function_name/params.description.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 1828571e-8fca-4e6a-851c-4c31c7c4c33f
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (11)
Cargo.tomlconnect/a2a/Cargo.tomlconnect/a2a/src/handler.rsconnect/a2a/src/main.rsconnect/a2a/src/types.rsconnect/mcp/Cargo.tomlconnect/mcp/src/handler.rsconnect/mcp/src/main.rsconnect/mcp/src/prompts.rsconnect/mcp/src/transport.rsconnect/mcp/src/worker_manager.rs
✅ Files skipped from review due to trivial changes (2)
- Cargo.toml
- connect/mcp/Cargo.toml
Comment on lines
+43
to
+50
| let iii = register_worker(&args.engine_url, InitOptions::default()); | ||
|
|
||
| handler::register(&iii, args.expose_all); | ||
|
|
||
| tracing::info!("A2A endpoints registered on engine port. Ctrl+C to stop."); | ||
| tokio::signal::ctrl_c().await?; | ||
|
|
||
| Ok(()) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Find the main.rs file and examine it
cat -n connect/a2a/src/main.rsRepository: iii-hq/iii
Length of output: 1682
🏁 Script executed:
# Search for III SDK shutdown usage patterns in the codebase
rg "shutdown" --type rust -B 2 -A 2Repository: iii-hq/iii
Length of output: 50366
🏁 Script executed:
# Check the iii SDK lib to understand the shutdown contract
find . -name "lib.rs" -path "*/iii/*" | head -5Repository: iii-hq/iii
Length of output: 88
🏁 Script executed:
# Look for register_worker usage patterns
rg "register_worker" --type rust -B 3 -A 3Repository: iii-hq/iii
Length of output: 50368
🏁 Script executed:
# Check the III SDK shutdown methods and documentation
sed -n '80,130p' sdk/packages/rust/iii/src/lib.rsRepository: iii-hq/iii
Length of output: 1444
🏁 Script executed:
# Also check the III struct for shutdown method signatures
rg "pub.*fn shutdown" sdk/packages/rust/iii/src/iii.rs -A 5Repository: iii-hq/iii
Length of output: 600
Call shutdown_async().await before exiting to flush telemetry.
register_worker() starts a background connection thread and telemetry collection. After ctrl_c(), the function returns immediately without shutting down these resources. Add iii.shutdown_async().await; before the Ok(()) to ensure graceful shutdown and telemetry flush.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/a2a/src/main.rs` around lines 43 - 50,
register_worker(&args.engine_url, InitOptions::default()) starts background work
and telemetry but the code returns immediately after Ctrl+C; call
iii.shutdown_async().await before returning to gracefully stop the background
connection and flush telemetry. Locate the variable iii (from register_worker)
and insert an awaited call to its shutdown_async() method just prior to
returning Ok(()) so shutdown_async().await runs after
tokio::signal::ctrl_c().await.
Contributor
Author
There was a problem hiding this comment.
The process exits on Ctrl+C. The iii-sdk worker disconnects automatically when the WebSocket drops. Explicit shutdown() is for clean telemetry flush (otel feature), which this binary doesn't enable.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 12 seconds before sending another message.
Contributor
Author
There was a problem hiding this comment.
Resolving — WebSocket auto-disconnects on process exit. Explicit shutdown only needed for otel flush.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 41 seconds before sending another message.
Comment on lines
+332
to
+341
| match self.iii.register_trigger(RegisterTriggerInput { | ||
| trigger_type: p.trigger_type.clone(), | ||
| function_id: p.function_id.clone(), | ||
| config: p.config, | ||
| }) { | ||
| Ok(trigger) => { | ||
| let id = uuid::Uuid::new_v4().to_string(); | ||
| self.triggers.lock().await.insert(id.clone(), trigger); | ||
| tool_json( | ||
| &json!({ "id": id, "trigger_type": p.trigger_type, "function_id": p.function_id }), |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C4 'pub struct TriggerInfo|pub struct Trigger|impl Trigger|fn unregister|register_trigger' sdk/packages/rust/iii/src
sed -n '321,362p' connect/mcp/src/handler.rsRepository: iii-hq/iii
Length of output: 18853
Return the engine-side trigger ID from register_trigger, not a session-local UUID.
Trigger (sdk/packages/rust/iii/src/triggers.rs) does not expose the engine-generated ID. The current implementation generates a new UUID locally and stores it in self.triggers, making unregister operations work only within the same session. Clients calling trigger_unregister in a different session cannot match the trigger by ID, and listing triggers from the engine returns IDs that don't correspond to the handler's internal mapping. This breaks cross-session trigger management.
Either modify Trigger to expose the engine-side id, or change register_trigger's return type to include it alongside the handle.
Also applies to: 357-362
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/mcp/src/handler.rs` around lines 332 - 341, The handler is returning
a locally-generated UUID instead of the engine-side trigger ID, which breaks
cross-session trigger management; update the code that calls
self.iii.register_trigger (and similarly the block at 357-362) to use the
engine-provided ID: either modify the Trigger type
(sdk/packages/rust/iii/src/triggers.rs) to expose its engine-side id field or
change register_trigger to return a tuple/struct that contains both the engine
id and the handle, then store the trigger handle in self.triggers keyed by the
engine id (not a new uuid) and return that engine id in the tool_json response
so unregister/list operations work across sessions (ensure you update usages of
register_trigger, self.triggers.insert, and the tool_json payload accordingly).
Contributor
Author
There was a problem hiding this comment.
Correct behavior. The iii-sdk Trigger struct has no id field — it only has unregister(). The engine generates the ID internally. We store the Trigger handle in a HashMap keyed by our generated UUID so trigger_unregister can look it up and call trigger.unregister(). Confirmed against SDK docs.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 10 seconds before sending another message.
Contributor
Author
There was a problem hiding this comment.
Resolving — confirmed against SDK docs. Trigger has no id field, our UUID-keyed HashMap is the correct pattern.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 39 seconds before sending another message.
Comment on lines
+16
to
+24
| loop { | ||
| while let Some(notification) = handler.take_notification().await { | ||
| let _ = writer.write_all(notification.as_bytes()).await; | ||
| let _ = writer.write_all(b"\n").await; | ||
| let _ = writer.flush().await; | ||
| } | ||
|
|
||
| let mut line = String::new(); | ||
| match reader.read_line(&mut line).await { |
Contributor
There was a problem hiding this comment.
Notifications are blocked behind the next stdin request.
After the initial drain, the loop waits in read_line().await, so any later notifications/tools/list_changed stays queued until the client sends another line. Idle clients will miss prompt tool-list updates unless this is driven with tokio::select! or a dedicated writer task.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/mcp/src/transport.rs` around lines 16 - 24, The current loop drains
handler.take_notification() but then blocks on reader.read_line(&mut line).await
so subsequent handler.take_notification() notifications (e.g.,
notifications/tools/list_changed) are delayed until the next stdin line; fix by
making the read and notification waits concurrent — either: (A) use
tokio::select! between a futures-aware notification future
(handler.take_notification()) and reader.read_line(&mut line) so you respond
immediately to whichever completes (on notification branch write via
writer.write_all/flush), or (B) move notification handling into a dedicated task
that owns writer and consumes notifications from handler, while the main task
only awaits read_line; update the loop to use the chosen approach and ensure
writer access is synchronized/owned by the notification task and that errors
from writer.write_all/flush are handled.
Contributor
Author
There was a problem hiding this comment.
Same as earlier comment — acceptable tradeoff. tools/list_changed fires rarely. MCP clients handle it on next interaction.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 9 seconds before sending another message.
Contributor
Author
There was a problem hiding this comment.
Resolving — acceptable for MCP. tools/list_changed is rare.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 38 seconds before sending another message.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (5)
connect/mcp/src/worker_manager.rs (2)
192-203:⚠️ Potential issue | 🔴 CriticalUse
description=in the generated Pythonregister_function()call.The template still emits a third positional argument. In the Python SDK,
descriptionis keyword-only, so generated Python workers will fail withTypeErrorbefore they register.🔧 Minimal fix
-iii.register_function('{}', handler, '{}') +iii.register_function('{}', handler, description='{}')Verify against the SDK signature with:
#!/bin/bash set -euo pipefail rg -n -A6 -B2 'def register_function\(' sdk/packages/python/iii/src/iii/iii.py sed -n '192,204p' connect/mcp/src/worker_manager.rs🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 192 - 203, The Python template currently emits a third positional argument to iii.register_function (see the format string using "iii.register_function('{}', handler, '{}')"), which fails because description is keyword-only; change the template to call iii.register_function('{}', handler, description='{}') and update the corresponding format placeholders/order so the description is passed as a keyword argument; keep the other parts (register_worker('{}'), Logger(), and the inserted handler block) unchanged.
65-89:⚠️ Potential issue | 🟡 MinorClean up temp dirs on failed creation paths.
create_worker()createstemp_dirbefore the unsupported-language branch, the file write, and the spawn. Any failure on those paths leaves orphanediii-*artifacts behind in the OS temp directory.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 65 - 89, create_worker currently creates temp_dir before generating/writing code and spawning the process, but any early return (unsupported language, write error, spawn error) leaves orphaned temp dirs; modify create_worker to remove the temp_dir on all failure paths: call tokio::fs::remove_dir_all(&temp_dir).await when returning Err (e.g., before returning from the unsupported-language branch, in each map_err closure for create_dir_all/write, and if spawn_worker returns Err), or refactor to capture the final Result and run a single cleanup on Err (using a match or a scope-guard/finally-style helper). Reference symbols: create_worker, temp_dir, generate_node_worker, generate_python_worker, spawn_worker.connect/mcp/src/handler.rs (3)
239-247:⚠️ Potential issue | 🔴 CriticalRe-check
mcp.exposebefore forwarding generic tool calls.Both stdio and HTTP turn user-supplied tool names into
function_ids and calliii.trigger(...)directly. Hidden functions stay callable by guessing the ID even when--expose-allis off.Also applies to: 439-446
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/handler.rs` around lines 239 - 247, The code builds a function_id from params.name and directly calls self.iii.trigger(...), which allows calling hidden tools by guessing IDs; before calling trigger (in the handler that constructs function_id and in the duplicate spot handling 439-446), check the mcp.expose policy (or the expose-all flag) to ensure the function_id is allowed to be invoked: determine if the target name is exposed via the mcp.expose list (or if expose-all is enabled) and only call self.iii.trigger when that check passes, otherwise return an error/permission denied response. Ensure you reference the same function_id/params.name logic and gate both locations where self.iii.trigger is invoked.
418-460:⚠️ Potential issue | 🟠 MajorDon't advertise HTTP tools/resources that
dispatch_http()can't serve.
tools/liststill includesbuiltin_tools(), but HTTPtools/callonly forwards names to the engine, soiii_worker_*/iii_trigger_*will fail over HTTP. The same path exposes resources viaresources/listwithout an HTTPresources/readimplementation.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/handler.rs` around lines 418 - 460, The response lists builtin tools and resources that cannot be served over HTTP; update the "tools/list" branch (the block using builtin_tools() and fns from iii.list_functions()) to filter out functions that dispatch_http() cannot call (e.g., exclude names like iii_worker_* and iii_trigger_* or apply a predicate that matches what dispatch_http supports) so tools/call (which just forwards names via iii.trigger) won't be advertised for HTTP, and either implement a matching "resources/read" handler or remove/stop advertising entries emitted by the "resources/list" branch that lack HTTP read support; adjust the tools/list filtering logic and the resources/list content accordingly so advertised endpoints reflect only actually supported HTTP operations.
338-343:⚠️ Potential issue | 🟠 MajorReturn the engine trigger ID, not a session-local UUID.
The ID you return/store here is unrelated to the engine trigger ID returned by
list_triggers().iii_trigger_unregisteronly works inside this handler instance, and clients cannot use IDs from the engine/resource view.Also applies to: 357-362
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/handler.rs` around lines 338 - 343, You're returning a session-local UUID instead of the engine's trigger ID; change the logic so you register the trigger with the engine, capture the engine's returned trigger_id, store the trigger in self.triggers keyed by that engine trigger_id (instead of uuid::Uuid::new_v4()), and return that engine trigger_id in the tool_json response; also update the corresponding insert/return in the similar block (the other occurrence around the second insert) and ensure iii_trigger_unregister and list_triggers() consumers work with the engine-returned ID rather than any session-local UUID.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connect/mcp/src/handler.rs`:
- Around line 185-234: The built-in trigger handlers (the match arms for
"iii_trigger_register", "iii_trigger_unregister", "iii_trigger_void", and
"iii_trigger_enqueue") currently accept arbitrary function_id values without
checking the MCP exposure gate; add a check after extracting fid (the result of
str_field or str_field_or) that verifies the function is exposed via the MCP
exposure mechanism (e.g., check mcp.expose or call the existing exposure-check
helper on self/mcp) and return Ok(tool_error("Function not exposed")) if it is
not exposed, then only construct and await the TriggerRequest
(self.iii.trigger(...)) when the exposure check passes—apply this change inside
the code paths that create TriggerRequest in these handlers (where fid is used)
and to the register/unregister implementations as well.
- Around line 386-394: The code calls iii.register_trigger(RegisterTriggerInput
{ ... }) and drops the returned Trigger handle immediately, losing runtime
control to unregister it; change the call to capture and store the returned
Trigger in a long-lived location (e.g., add a field like http_trigger:
Option<Trigger> on the owning struct such as Handler or a static/once_cell if
global) instead of discarding it, assign Some(trigger) after successful
registration, and update initialization/cleanup logic to use that stored handle
for unregistering when needed.
---
Duplicate comments:
In `@connect/mcp/src/handler.rs`:
- Around line 239-247: The code builds a function_id from params.name and
directly calls self.iii.trigger(...), which allows calling hidden tools by
guessing IDs; before calling trigger (in the handler that constructs function_id
and in the duplicate spot handling 439-446), check the mcp.expose policy (or the
expose-all flag) to ensure the function_id is allowed to be invoked: determine
if the target name is exposed via the mcp.expose list (or if expose-all is
enabled) and only call self.iii.trigger when that check passes, otherwise return
an error/permission denied response. Ensure you reference the same
function_id/params.name logic and gate both locations where self.iii.trigger is
invoked.
- Around line 418-460: The response lists builtin tools and resources that
cannot be served over HTTP; update the "tools/list" branch (the block using
builtin_tools() and fns from iii.list_functions()) to filter out functions that
dispatch_http() cannot call (e.g., exclude names like iii_worker_* and
iii_trigger_* or apply a predicate that matches what dispatch_http supports) so
tools/call (which just forwards names via iii.trigger) won't be advertised for
HTTP, and either implement a matching "resources/read" handler or remove/stop
advertising entries emitted by the "resources/list" branch that lack HTTP read
support; adjust the tools/list filtering logic and the resources/list content
accordingly so advertised endpoints reflect only actually supported HTTP
operations.
- Around line 338-343: You're returning a session-local UUID instead of the
engine's trigger ID; change the logic so you register the trigger with the
engine, capture the engine's returned trigger_id, store the trigger in
self.triggers keyed by that engine trigger_id (instead of uuid::Uuid::new_v4()),
and return that engine trigger_id in the tool_json response; also update the
corresponding insert/return in the similar block (the other occurrence around
the second insert) and ensure iii_trigger_unregister and list_triggers()
consumers work with the engine-returned ID rather than any session-local UUID.
In `@connect/mcp/src/worker_manager.rs`:
- Around line 192-203: The Python template currently emits a third positional
argument to iii.register_function (see the format string using
"iii.register_function('{}', handler, '{}')"), which fails because description
is keyword-only; change the template to call iii.register_function('{}',
handler, description='{}') and update the corresponding format
placeholders/order so the description is passed as a keyword argument; keep the
other parts (register_worker('{}'), Logger(), and the inserted handler block)
unchanged.
- Around line 65-89: create_worker currently creates temp_dir before
generating/writing code and spawning the process, but any early return
(unsupported language, write error, spawn error) leaves orphaned temp dirs;
modify create_worker to remove the temp_dir on all failure paths: call
tokio::fs::remove_dir_all(&temp_dir).await when returning Err (e.g., before
returning from the unsupported-language branch, in each map_err closure for
create_dir_all/write, and if spawn_worker returns Err), or refactor to capture
the final Result and run a single cleanup on Err (using a match or a
scope-guard/finally-style helper). Reference symbols: create_worker, temp_dir,
generate_node_worker, generate_python_worker, spawn_worker.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 42af6b74-eb75-49d8-8042-bf93f89cb2b5
📒 Files selected for processing (4)
connect/a2a/src/handler.rsconnect/mcp/src/handler.rsconnect/mcp/src/main.rsconnect/mcp/src/worker_manager.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- connect/mcp/src/main.rs
- connect/a2a/src/handler.rs
Comment on lines
+185
to
+234
| "iii_trigger_register" => return Ok(self.trigger_register(params.arguments).await), | ||
| "iii_trigger_unregister" => return Ok(self.trigger_unregister(params.arguments).await), | ||
| "iii_trigger_void" => { | ||
| let fid = str_field(¶ms.arguments, "function_id"); | ||
| if fid.is_empty() { | ||
| return Ok(tool_error("Missing required field: function_id")); | ||
| } | ||
| let payload = params | ||
| .arguments | ||
| .get("payload") | ||
| .cloned() | ||
| .unwrap_or(json!({})); | ||
| return match self | ||
| .iii | ||
| .trigger(TriggerRequest { | ||
| function_id: fid.clone(), | ||
| payload, | ||
| action: Some(TriggerAction::Void), | ||
| timeout_ms: None, | ||
| }) | ||
| .await | ||
| { | ||
| Ok(_) => Ok(tool_result(&format!("Triggered (void): {}", fid))), | ||
| Err(e) => Ok(tool_error(&format!("Error: {}", e))), | ||
| }; | ||
| } | ||
| "iii_trigger_enqueue" => { | ||
| let fid = str_field(¶ms.arguments, "function_id"); | ||
| if fid.is_empty() { | ||
| return Ok(tool_error("Missing required field: function_id")); | ||
| } | ||
| let payload = params | ||
| .arguments | ||
| .get("payload") | ||
| .cloned() | ||
| .unwrap_or(json!({})); | ||
| let queue = str_field_or(¶ms.arguments, "queue", "default"); | ||
| return match self | ||
| .iii | ||
| .trigger(TriggerRequest { | ||
| function_id: fid, | ||
| payload, | ||
| action: Some(TriggerAction::Enqueue { queue }), | ||
| timeout_ms: None, | ||
| }) | ||
| .await | ||
| { | ||
| Ok(r) => Ok(tool_json(&r)), | ||
| Err(e) => Ok(tool_error(&format!("Error: {}", e))), | ||
| }; |
Contributor
There was a problem hiding this comment.
Apply the same exposure gate to the built-in trigger tools.
iii_trigger_void, iii_trigger_enqueue, and iii_trigger_register all accept arbitrary function_ids without checking mcp.expose. Even if the generic call path is fixed, these built-ins still let clients invoke or wire triggers to hidden functions.
Also applies to: 321-336
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/mcp/src/handler.rs` around lines 185 - 234, The built-in trigger
handlers (the match arms for "iii_trigger_register", "iii_trigger_unregister",
"iii_trigger_void", and "iii_trigger_enqueue") currently accept arbitrary
function_id values without checking the MCP exposure gate; add a check after
extracting fid (the result of str_field or str_field_or) that verifies the
function is exposed via the MCP exposure mechanism (e.g., check mcp.expose or
call the existing exposure-check helper on self/mcp) and return
Ok(tool_error("Function not exposed")) if it is not exposed, then only construct
and await the TriggerRequest (self.iii.trigger(...)) when the exposure check
passes—apply this change inside the code paths that create TriggerRequest in
these handlers (where fid is used) and to the register/unregister
implementations as well.
Contributor
Author
There was a problem hiding this comment.
The builtin tools (iii_trigger_void, iii_trigger_enqueue, iii_trigger_register) are MCP-level management tools, not proxies to hidden functions. They're always visible because they ARE the MCP tooling — an agent uses them to interact with the engine directly. The function_id they accept is validated by the engine at invocation time. When RBAC lands, the engine rejects unauthorized trigger() calls regardless of what the MCP layer exposes. The metadata filter is for discovery noise reduction, not access control.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 1 minutes and 18 seconds before sending another message.
Comment on lines
+386
to
+394
| if let Err(e) = iii.register_trigger(RegisterTriggerInput { | ||
| trigger_type: "http".to_string(), | ||
| function_id: "mcp::handler".to_string(), | ||
| config: json!({ "api_path": "/mcp", "http_method": "POST" }), | ||
| }) { | ||
| tracing::error!(error = %e, "Failed to register MCP HTTP trigger"); | ||
| } else { | ||
| tracing::info!("MCP Streamable HTTP registered: POST /mcp"); | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C4 'pub struct Trigger|impl Drop for Trigger|fn unregister\(' sdk/packages/rust/iii/src/triggers.rs
sed -n '386,394p' connect/mcp/src/handler.rsRepository: iii-hq/iii
Length of output: 1248
Store the Trigger handle if the MCP HTTP endpoint lifecycle needs to be managed.
The returned Trigger handle is dropped immediately without being stored. While the HTTP trigger will persist (the SDK does not auto-unregister on Drop), the application loses the ability to unregister this endpoint at runtime. If this is intentional (fire-and-forget registration for the application lifetime), it's acceptable; otherwise, the Trigger should be stored in a struct field or static to maintain control over its lifecycle.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connect/mcp/src/handler.rs` around lines 386 - 394, The code calls
iii.register_trigger(RegisterTriggerInput { ... }) and drops the returned
Trigger handle immediately, losing runtime control to unregister it; change the
call to capture and store the returned Trigger in a long-lived location (e.g.,
add a field like http_trigger: Option<Trigger> on the owning struct such as
Handler or a static/once_cell if global) instead of discarding it, assign
Some(trigger) after successful registration, and update initialization/cleanup
logic to use that stored handle for unregistering when needed.
Contributor
Author
There was a problem hiding this comment.
Intentional. The MCP HTTP trigger lives for the entire process lifetime — it's registered at startup and never unregistered. Storing the handle would add state management for a trigger that's never removed. The SDK doesn't auto-unregister on Drop (as the analysis confirmed), so the trigger persists correctly.
Contributor
There was a problem hiding this comment.
Rate Limit Exceeded
@rohitg00 have exceeded the limit for the number of chat messages per hour. Please wait 0 minutes and 12 seconds before sending another message.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (3)
connect/a2a/src/handler.rs (1)
395-417:⚠️ Potential issue | 🟠 MajorDon't mask
state::listfailures as an empty task list.An engine/storage error is currently indistinguishable from “no tasks exist”, which will mislead clients and hide backend outages. Return a JSON-RPC error here and log the actual failure instead of synthesizing
{ "tasks": [] }.🔧 Minimal fix
- Err(_) => A2AResponse::success(id, json!({ "tasks": [] })), + Err(e) => { + tracing::error!(error = %e, "Failed to list A2A tasks"); + A2AResponse::error(id, -32603, "Failed to list tasks") + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/a2a/src/handler.rs` around lines 395 - 417, The handler handle_list currently treats any Err from iii.trigger(TriggerRequest { function_id: "state::list", ... }) as a successful empty list; change this so that on Err(e) you log the error (including e) and return an A2AResponse JSON-RPC error instead of A2AResponse::success(..., { "tasks":[] }); specifically, capture the error from the trigger call, emit a clear log line (e.g., via your existing logger or tracing::error!) mentioning "state::list failed" and the error, and construct and return an A2AResponse error variant carrying an appropriate JSON-RPC error object (with message and/or code) rather than synthesizing an empty tasks array.connect/mcp/src/worker_manager.rs (2)
199-225:⚠️ Potential issue | 🔴 CriticalPass
descriptionas a keyword in the generated Python call.Line 209 still emits
iii.register_function(id, handler, description)positionally. The Python SDK only accepts two positional arguments here, so generated Python workers will fail withTypeErroron startup.🐍 Minimal fix
-iii.register_function('{}', handler, '{}') +iii.register_function('{}', handler, description='{}')Read-only verification. Expected result: the SDK definition shows
descriptionafter the keyword-only separator, while this template still passes it positionally.#!/bin/bash rg -n -C3 'def register_function\(' sdk/packages/python/iii/src/iii/iii.py nl -ba connect/mcp/src/worker_manager.rs | sed -n '204,210p'🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 199 - 225, The generated Python template calls iii.register_function positionally with description, causing a TypeError; update the format string in worker_manager.rs so the generated line uses a keyword argument for description (e.g., call iii.register_function(function_name, handler, description=description)) by changing the register_function placeholder from a third positional argument to a keyword form and adjust the format! placeholders accordingly (refer to the format! invocation and the symbols self.engine_url, params.code, function_name, description to locate and reorder the arguments).
65-85:⚠️ Potential issue | 🟡 MinorClean up
temp_diron the remaining early returns.The spawn-failure path is fixed, but Line 79 and Lines 83-85 still return after creating
temp_dir. A stream of unsupported-language or write-failure requests will keep leaking temp directories.🧹 Minimal fix
let (file_name, code) = match params.language.as_str() { "node" | "javascript" | "js" => { let code = self.generate_node_worker(¶ms); ("index.mjs", code) } "python" | "py" => { let code = self.generate_python_worker(¶ms); ("main.py", code) } - _ => return Err(format!("Unsupported language: {}", params.language)), + _ => { + let _ = tokio::fs::remove_dir_all(&temp_dir).await; + return Err(format!("Unsupported language: {}", params.language)); + } }; let file_path = temp_dir.join(file_name); - tokio::fs::write(&file_path, &code) - .await - .map_err(|e| format!("Failed to write worker file: {}", e))?; + if let Err(e) = tokio::fs::write(&file_path, &code).await { + let _ = tokio::fs::remove_dir_all(&temp_dir).await; + return Err(format!("Failed to write worker file: {}", e)); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 65 - 85, The code creates temp_dir (temp_dir variable) then returns early on unsupported language or write errors, leaking the directory; ensure temp_dir is removed on all early-return paths by calling tokio::fs::remove_dir_all(&temp_dir).await.map_err(|e| format!("Failed to cleanup temp dir: {}", e))? before each Err return (the Unsupported language branch and the tokio::fs::write error path), or refactor to use an RAII guard (a small TempDirDrop struct with Drop that calls tokio::spawn to remove_dir_all) so that generate_node_worker/generate_python_worker and the file_path write both run safely and temp_dir is always cleaned up even on errors.
🧹 Nitpick comments (1)
connect/mcp/src/worker_manager.rs (1)
127-137: Drop the worker-map lock before the async cleanup.
stop_worker()holds the sharedworkersmutex acrosschild.kill().awaitandremove_dir_all().await, so one slow stop blocks every other create/stop request on this handler instance.🔓 Small refactor
pub async fn stop_worker(&self, params: WorkerStopParams) -> Result<WorkerStopResult, String> { - let mut workers = self.workers.lock().await; - - if let Some((info, mut child)) = workers.remove(¶ms.id) { + let worker = { + let mut workers = self.workers.lock().await; + workers.remove(¶ms.id) + }; + + if let Some((info, mut child)) = worker { if let Err(e) = child.kill().await { tracing::warn!(worker_id = %params.id, error = %e, "Failed to kill worker process"); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connect/mcp/src/worker_manager.rs` around lines 127 - 137, stop_worker currently holds the workers mutex across async awaits (child.kill().await and tokio::fs::remove_dir_all), blocking other handlers; fix it by removing the entry while holding the lock then immediately dropping the lock before performing async cleanup. Concretely: inside stop_worker, call self.workers.lock().await, remove the map entry with workers.remove(¶ms.id) and store the returned (info, child) in local variables (matching WorkerStopParams/WorkerStopResult), then drop the mutex guard (or let the scope end / call drop(workers)) before awaiting child.kill() and tokio::fs::remove_dir_all(&info.temp_dir); preserve the existing tracing::warn! error handling after each await.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connect/a2a/src/handler.rs`:
- Around line 282-308: Guard the caller-supplied task_id before using it as the
state key: when handling params.message.task_id in Task construction, treat
empty/blank values as invalid and generate a new UUID instead, and before
calling store_task check the backing store for an existing task with that id
(e.g., via the existing load/get helper or a new fetch by
"a2a:tasks/<task_id>"); if an existing task is found either merge the incoming
message into the existing Task (append to history and merge artifacts/metadata)
and update status, or reject the duplicate id and return an error to the
caller—do not blindly overwrite. Ensure you reference and use the symbols Task,
params.message.task_id, task_id, store_task (and the store load/get helper) so
callers with follow-up messages can be handled via load+merge or explicit
rejection.
- Around line 451-476: Trim the extracted text at the top of resolve_function
(i.e., set text = text.trim()) and when splitting the "function_id" from its
payload use a whitespace-aware splitter (e.g., text.splitn(2,
char::is_whitespace()) or equivalent) instead of splitn(2, ' '); keep the
existing payload parsing logic (serde_json::from_str(...).unwrap_or(json!({
"input": parts[1] }))) and return the same tuple shape so indented or
newline-separated inputs like "foo::bar\n{...}" resolve correctly.
---
Duplicate comments:
In `@connect/a2a/src/handler.rs`:
- Around line 395-417: The handler handle_list currently treats any Err from
iii.trigger(TriggerRequest { function_id: "state::list", ... }) as a successful
empty list; change this so that on Err(e) you log the error (including e) and
return an A2AResponse JSON-RPC error instead of A2AResponse::success(..., {
"tasks":[] }); specifically, capture the error from the trigger call, emit a
clear log line (e.g., via your existing logger or tracing::error!) mentioning
"state::list failed" and the error, and construct and return an A2AResponse
error variant carrying an appropriate JSON-RPC error object (with message and/or
code) rather than synthesizing an empty tasks array.
In `@connect/mcp/src/worker_manager.rs`:
- Around line 199-225: The generated Python template calls iii.register_function
positionally with description, causing a TypeError; update the format string in
worker_manager.rs so the generated line uses a keyword argument for description
(e.g., call iii.register_function(function_name, handler,
description=description)) by changing the register_function placeholder from a
third positional argument to a keyword form and adjust the format! placeholders
accordingly (refer to the format! invocation and the symbols self.engine_url,
params.code, function_name, description to locate and reorder the arguments).
- Around line 65-85: The code creates temp_dir (temp_dir variable) then returns
early on unsupported language or write errors, leaking the directory; ensure
temp_dir is removed on all early-return paths by calling
tokio::fs::remove_dir_all(&temp_dir).await.map_err(|e| format!("Failed to
cleanup temp dir: {}", e))? before each Err return (the Unsupported language
branch and the tokio::fs::write error path), or refactor to use an RAII guard (a
small TempDirDrop struct with Drop that calls tokio::spawn to remove_dir_all) so
that generate_node_worker/generate_python_worker and the file_path write both
run safely and temp_dir is always cleaned up even on errors.
---
Nitpick comments:
In `@connect/mcp/src/worker_manager.rs`:
- Around line 127-137: stop_worker currently holds the workers mutex across
async awaits (child.kill().await and tokio::fs::remove_dir_all), blocking other
handlers; fix it by removing the entry while holding the lock then immediately
dropping the lock before performing async cleanup. Concretely: inside
stop_worker, call self.workers.lock().await, remove the map entry with
workers.remove(¶ms.id) and store the returned (info, child) in local
variables (matching WorkerStopParams/WorkerStopResult), then drop the mutex
guard (or let the scope end / call drop(workers)) before awaiting child.kill()
and tokio::fs::remove_dir_all(&info.temp_dir); preserve the existing
tracing::warn! error handling after each await.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 25c69e1b-8f4d-458f-bd16-74e0791a0b3c
📒 Files selected for processing (2)
connect/a2a/src/handler.rsconnect/mcp/src/worker_manager.rs
…TP, eliminate global static
… crates.io publishing
rohitg00
added a commit
to iii-hq/workers
that referenced
this pull request
Mar 30, 2026
Move MCP and A2A protocol workers from iii-hq/iii#1374 into the standalone workers repo. Each is its own Rust crate with crates.io iii-sdk dependency (no monorepo path refs). - iii-mcp: MCP protocol worker (stdio + Streamable HTTP) - iii-a2a: A2A protocol worker (agent card + JSON-RPC) - Updated registry/index.json with both workers - Updated README with usage docs
6 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Two standalone protocol workers under
workers/. Each is its own crate with its own binary. No shared library.iii-mcp— MCP protocol worker (stdio + Streamable HTTP)iii-a2a— A2A protocol worker (HTTP on engine port)Follows the worker pattern:
iii worker add mcp/iii worker add a2a.Usage
Metadata filtering
Functions need metadata tags to appear as tools/skills:
Without the tag, functions stay hidden from agents.
--expose-alldisables filtering.Structure
Publishing
Both crates use
iii-sdk = { version = "0.10.0", path = "..." }— path for monorepo builds, version for crates.io publishing. Keywords and categories set for discoverability.Commits
feat: add iii-mcp and iii-a2a protocol workers— initial implementationfix: address CodeRabbit review— expose_all in HTTP, template escaping, A2A error envelopefix: fail fast on missing function_id, clean up temp dir on spawn failurefix: schema/impl consistency, safe serialize, remove builtins from HTTP, eliminate global staticrefactor: move to workers/ dir, add version+path dual depTest plan
cargo clippy -p iii-mcp -p iii-a2a -- -D warningsclean