bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add session and async job mcp tools

Browse Source
This commit is contained in:
a.tolmachev
2026-04-06 11:35:37 +03:00
parent bd2c6d4f48
commit fdd0a45124
5 changed files with 1578 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TASKS.md
+7 -7
View File
@@ -2,20 +2,20 @@
## Current
### `feat/grpc-server-streaming-adapter`
### `feat/session-and-job-tools`
Status: completed
DoD:
- gRPC adapter supports bounded server-stream collection for `window` mode
- streamed protobuf messages are decoded into normalized JSON items
- adapter respects item/window limits and upstream timeout
- runtime dispatches gRPC `window` mode through streaming adapter path
- local gRPC upstream tests cover success, timeout and malformed method kind handling
- session and async_job tool families are published in the MCP tool catalog
- generated `start/poll/stop` and `start/status/result/cancel` tool calls work through JSON-RPC
- stream sessions and async jobs are persisted with explicit status transitions
- session and async_job tool calls are logged through observability
- MCP integration tests cover `session start -> poll -> stop` and `async job start -> status -> result`
## Next
- `feat/grpc-server-streaming-adapter`
- `feat/streaming-ui-config`
## Backlog
apps/mcp-server/src/app.rs
+1136 -109
View File
File diff suppressed because it is too large Load Diff
apps/mcp-server/src/catalog.rs
-21
View File
@@ -25,7 +25,6 @@ struct CatalogKey {
struct CachedCatalog {
loaded_at: Option<Instant>,
tools: Vec<PublishedAgentTool>,
tools_by_name: HashMap<String, PublishedAgentTool>,
}
impl PublishedToolCatalog {
@@ -50,20 +49,6 @@ impl PublishedToolCatalog {
.unwrap_or_default())
}
pub async fn get_tool(
&self,
workspace_slug: &str,
agent_slug: &str,
tool_name: &str,
) -> Result<Option<PublishedAgentTool>, RegistryError> {
self.refresh_if_stale(workspace_slug, agent_slug).await?;
let guard = self.cached.read().await;
Ok(guard
.get(&CatalogKey::new(workspace_slug, agent_slug))
.and_then(|entry| entry.tools_by_name.get(tool_name))
.cloned())
}
async fn refresh_if_stale(
&self,
workspace_slug: &str,
@@ -92,11 +77,6 @@ impl PublishedToolCatalog {
Err(RegistryError::PublishedAgentNotFound { .. }) => Vec::new(),
Err(error) => return Err(error),
};
let tools_by_name = tools
.iter()
.cloned()
.map(|tool| (tool.tool_name.clone(), tool))
.collect::<HashMap<_, _>>();
let mut guard = self.cached.write().await;
let previous_count = guard
.get(&key)
@@ -108,7 +88,6 @@ impl PublishedToolCatalog {
CachedCatalog {
loaded_at: Some(Instant::now()),
tools,
tools_by_name,
},
);
apps/mcp-server/src/main.rs
+401 -5
View File
@@ -52,10 +52,11 @@ mod tests {
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_adapter_grpc::test_support as grpc_test_support;
use crank_core::{
Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, DescriptorId,
ExecutionConfig, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod, Operation,
OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AggregationMode,
DescriptorId, ExecutionConfig, ExecutionMode, GraphqlOperationType, GraphqlTarget,
GrpcTarget, HttpMethod, Operation, OperationId, OperationStatus, PlatformApiKey,
PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, Protocol, RestTarget,
StreamingConfig, Target, ToolDescription, ToolFamilyConfig, TransportBehavior, WorkspaceId,
};
use crank_mapping::{MappingRule, MappingSet};
use crank_registry::{
@@ -67,6 +68,7 @@ mod tests {
use sha2::{Digest, Sha256};
use sqlx::{Executor, postgres::PgPoolOptions};
use tokio::net::TcpListener;
use tokio::time::sleep;
use crate::app::build_app;
@@ -821,6 +823,297 @@ mod tests {
assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
}
#[tokio::test]
async fn exposes_and_runs_session_tool_family_via_mcp() {
let registry = test_registry().await;
let server_addr = grpc_test_support::spawn_unary_echo_server().await;
let operation = test_grpc_session_operation(&server_addr, "echo_stream_session");
registry
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
published_by: Some("alice"),
})
.await
.unwrap();
publish_agent_for_operation(&registry, &operation, "sales-session").await;
let api_key = create_platform_api_key(
&registry,
"mcp-session",
&[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
)
.await;
let base_url = spawn_mcp_server(build_app(
registry.clone(),
Duration::from_millis(0),
Some("https://crank.example.com".to_owned()),
))
.await;
let client = reqwest::Client::new();
let mcp_url = agent_mcp_url(&base_url, "sales-session");
let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
let tools = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}),
)
.await;
let tool_names = tools["result"]["tools"]
.as_array()
.unwrap()
.iter()
.map(|tool| tool["name"].as_str().unwrap().to_owned())
.collect::<Vec<_>>();
assert!(!tool_names.contains(&operation.name));
assert!(tool_names.contains(&"echo_stream_session_start".to_owned()));
assert!(tool_names.contains(&"echo_stream_session_poll".to_owned()));
assert!(tool_names.contains(&"echo_stream_session_stop".to_owned()));
let start_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "echo_stream_session_start",
"arguments": {
"message": "hello"
}
}
}),
)
.await;
let session_id = start_response["result"]["structuredContent"]["session_id"]
.as_str()
.unwrap()
.to_owned();
assert_eq!(
start_response["result"]["structuredContent"]["status"],
json!("running")
);
let poll_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "echo_stream_session_poll",
"arguments": {
"session_id": session_id
}
}
}),
)
.await;
assert_eq!(
poll_response["result"]["structuredContent"]["session_id"],
json!(session_id)
);
assert!(
poll_response["result"]["structuredContent"]["items"]
.as_array()
.is_some_and(|items| !items.is_empty())
);
let stop_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "echo_stream_session_stop",
"arguments": {
"session_id": session_id
}
}
}),
)
.await;
assert_eq!(
stop_response["result"]["structuredContent"],
json!({
"session_id": session_id,
"status": "stopped"
})
);
}
#[tokio::test]
async fn exposes_and_runs_async_job_tool_family_via_mcp() {
let registry = test_registry().await;
let upstream_base_url = spawn_upstream_server().await;
let operation = test_rest_async_job_operation(&upstream_base_url, "crm_async_create_lead");
registry
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
published_by: Some("alice"),
})
.await
.unwrap();
publish_agent_for_operation(&registry, &operation, "sales-async").await;
let api_key = create_platform_api_key(
&registry,
"mcp-async",
&[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
)
.await;
let base_url = spawn_mcp_server(build_app(
registry.clone(),
Duration::from_millis(0),
Some("https://crank.example.com".to_owned()),
))
.await;
let client = reqwest::Client::new();
let mcp_url = agent_mcp_url(&base_url, "sales-async");
let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
let tools = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}),
)
.await;
let tool_names = tools["result"]["tools"]
.as_array()
.unwrap()
.iter()
.map(|tool| tool["name"].as_str().unwrap().to_owned())
.collect::<Vec<_>>();
assert!(!tool_names.contains(&operation.name));
assert!(tool_names.contains(&"crm_async_create_lead_start".to_owned()));
assert!(tool_names.contains(&"crm_async_create_lead_status".to_owned()));
assert!(tool_names.contains(&"crm_async_create_lead_result".to_owned()));
assert!(tool_names.contains(&"crm_async_create_lead_cancel".to_owned()));
let start_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "crm_async_create_lead_start",
"arguments": {
"email": "user@example.com"
}
}
}),
)
.await;
let job_id = start_response["result"]["structuredContent"]["job_id"]
.as_str()
.unwrap()
.to_owned();
assert_eq!(
start_response["result"]["structuredContent"]["status"],
json!("running")
);
let mut status_response = Value::Null;
for _ in 0..20 {
status_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "crm_async_create_lead_status",
"arguments": {
"job_id": job_id
}
}
}),
)
.await;
if status_response["result"]["structuredContent"]["status"] == json!("completed") {
break;
}
sleep(Duration::from_millis(25)).await;
}
assert_eq!(
status_response["result"]["structuredContent"]["status"],
json!("completed")
);
let result_response = post_jsonrpc(
&client,
&mcp_url,
&api_key,
Some(&initialized_session),
json!({
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "crm_async_create_lead_result",
"arguments": {
"job_id": job_id
}
}
}),
)
.await;
assert_eq!(
result_response["result"]["structuredContent"],
json!({ "id": "lead_123" })
);
}
async fn initialize_session(client: &reqwest::Client, mcp_url: &str, api_key: &str) -> String {
let initialize_response = client
.post(mcp_url)
@@ -926,7 +1219,9 @@ mod tests {
}
async fn spawn_upstream_server() -> String {
let app = Router::new().route("/crm/leads", post(create_lead));
let app = Router::new()
.route("/crm/leads", post(create_lead))
.route("/crm/slow-leads", post(create_slow_lead));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
@@ -1039,6 +1334,15 @@ mod tests {
}))
}
async fn create_slow_lead(Json(payload): Json<Value>) -> Json<Value> {
sleep(Duration::from_millis(250)).await;
Json(json!({
"id": "lead_123",
"email": payload["email"]
}))
}
async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
let email = payload
.get("variables")
@@ -1274,6 +1578,98 @@ mod tests {
}
}
fn test_grpc_session_operation(server_addr: &str, name: &str) -> Operation<Schema, MappingSet> {
let mut operation = test_grpc_operation(server_addr, name);
operation.target = Target::Grpc(GrpcTarget {
server_addr: server_addr.to_owned(),
package: "echo".to_owned(),
service: "EchoService".to_owned(),
method: "ServerEcho".to_owned(),
descriptor_ref: DescriptorId::new("desc_echo"),
descriptor_set_b64: grpc_test_support::descriptor_set_b64(),
});
operation.display_name = "Server Echo Session".to_owned();
operation.tool_description = ToolDescription {
title: "Server Echo Session".to_owned(),
description: "Streams echo messages through a bounded session".to_owned(),
tags: vec!["grpc".to_owned(), "streaming".to_owned()],
examples: Vec::new(),
};
operation.execution_config.streaming = Some(StreamingConfig {
mode: ExecutionMode::Session,
transport_behavior: TransportBehavior::ServerStream,
window_duration_ms: Some(1_000),
poll_interval_ms: Some(250),
upstream_timeout_ms: Some(1_000),
idle_timeout_ms: Some(5_000),
max_session_lifetime_ms: Some(60_000),
max_items: Some(1),
max_bytes: Some(16 * 1024),
aggregation_mode: AggregationMode::SummaryPlusSamples,
summary_path: None,
items_path: Some("$.items".to_owned()),
cursor_path: None,
status_path: None,
done_path: Some("$.done".to_owned()),
redacted_paths: Vec::new(),
truncate_item_fields: false,
max_field_length: None,
drop_duplicates: false,
sampling_rate: None,
tool_family: ToolFamilyConfig {
start_tool_name: Some(format!("{name}_start")),
poll_tool_name: Some(format!("{name}_poll")),
stop_tool_name: Some(format!("{name}_stop")),
status_tool_name: None,
result_tool_name: None,
cancel_tool_name: None,
},
});
operation
}
fn test_rest_async_job_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
let mut operation = test_operation(base_url, name);
operation.display_name = "Create Lead Async".to_owned();
operation.tool_description = ToolDescription {
title: "Create Lead Async".to_owned(),
description: "Creates a CRM lead through an async job tool family".to_owned(),
tags: vec!["rest".to_owned(), "async_job".to_owned()],
examples: Vec::new(),
};
operation.execution_config.streaming = Some(StreamingConfig {
mode: ExecutionMode::AsyncJob,
transport_behavior: TransportBehavior::DeferredResult,
window_duration_ms: None,
poll_interval_ms: Some(250),
upstream_timeout_ms: Some(2_000),
idle_timeout_ms: None,
max_session_lifetime_ms: Some(300_000),
max_items: None,
max_bytes: Some(16 * 1024),
aggregation_mode: AggregationMode::SummaryOnly,
summary_path: None,
items_path: None,
cursor_path: None,
status_path: None,
done_path: None,
redacted_paths: Vec::new(),
truncate_item_fields: false,
max_field_length: None,
drop_duplicates: false,
sampling_rate: None,
tool_family: ToolFamilyConfig {
start_tool_name: Some(format!("{name}_start")),
poll_tool_name: None,
stop_tool_name: None,
status_tool_name: Some(format!("{name}_status")),
result_tool_name: Some(format!("{name}_result")),
cancel_tool_name: Some(format!("{name}_cancel")),
},
});
operation
}
fn object_schema(field_name: &str) -> Schema {
Schema {
kind: SchemaKind::Object,
crates/crank-runtime/src/executor.rs
+34
View File
@@ -74,6 +74,40 @@ impl RuntimeExecutor {
crate::aggregation::collect_window_result(&adapter_response.body, streaming)
}
pub async fn execute_session_seed(
&self,
operation: &RuntimeOperation,
input: &Value,
) -> Result<WindowExecutionResult, RuntimeError> {
let Some(streaming) = operation.execution_config.streaming.as_ref() else {
return Err(RuntimeError::MissingStreamingConfig {
operation_id: operation.operation_id.as_str().to_owned(),
});
};
if streaming.mode != ExecutionMode::Session {
return Err(RuntimeError::UnsupportedExecutionMode {
operation_id: operation.operation_id.as_str().to_owned(),
mode: streaming.mode,
});
}
let batch_size = streaming.max_items.unwrap_or(10).max(1);
let seed_limit = batch_size.saturating_mul(4);
let mut seeded_operation = operation.clone();
if let Some(config) = seeded_operation.execution_config.streaming.as_mut() {
config.mode = ExecutionMode::Window;
config.window_duration_ms = config
.window_duration_ms
.or(config.poll_interval_ms)
.or(config.upstream_timeout_ms)
.or(Some(operation.execution_config.timeout_ms));
config.max_items = Some(seed_limit);
}
self.execute_window(&seeded_operation, input).await
}
pub fn prepare_request(
&self,
operation: &RuntimeOperation,