use std::sync::atomic::{AtomicUsize, Ordering};
use async_trait::async_trait;
use supercode_harness::mcp::{McpClient, McpTool};
use supercode_harness::permissions::{
ApprovalOutcome, ApprovalRequest, PermissionsApprovalHandler,
};
use supercode_harness::tools::Tool;
use supercode_harness::{
Agent, ApprovalPolicy, ChatMessage, ChatRequest, Config, FunctionCall, Provider, Role,
ToolCall, Usage,
};
fn tmp() -> std::path::PathBuf {
static N: AtomicUsize = AtomicUsize::new(0);
let d = std::env::temp_dir().join(format!(
"sc-mcp-perm-{}-{}",
std::process::id(),
N.fetch_add(1, Ordering::SeqCst)
));
std::fs::create_dir_all(&d).unwrap();
d
}
fn have_python3() -> bool {
std::process::Command::new("python3")
.arg("--version")
.output()
.is_ok()
}
const ECHO_SERVER: &str = r#"
import sys, json
def send(o): sys.stdout.write(json.dumps(o)+"\n"); sys.stdout.flush()
for line in sys.stdin:
line=line.strip()
if not line: continue
msg=json.loads(line)
m=msg.get("method"); i=msg.get("id")
if m=="initialize":
send({"jsonrpc":"2.0","id":i,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"fake"}}})
elif m=="tools/list":
send({"jsonrpc":"2.0","id":i,"result":{"tools":[
{"name":"delete_everything","description":"dangerous","inputSchema":{"type":"object"}}
]}})
elif m=="tools/call":
send({"jsonrpc":"2.0","id":i,"result":{"content":[{"type":"text","text":"deleted"}],"isError":False}})
else:
if i is not None:
send({"jsonrpc":"2.0","id":i,"error":{"code":-32601,"message":"unknown"}})
"#;
async fn agent_with_mcp_tool(config: Config, provider: Box<dyn Provider>) -> Option<Agent> {
if !have_python3() {
eprintln!("skipping: no python3");
return None;
}
let dir = tmp();
let script = dir.join("server.py");
std::fs::write(&script, ECHO_SERVER).unwrap();
let client = McpClient::connect("python3", &[script.to_str().unwrap()], &Default::default())
.await
.unwrap();
let tools = McpTool::from_client("fake", client).await.unwrap();
assert_eq!(tools[0].name(), "mcp__fake__delete_everything");
let mut agent = Agent::with_provider(config, provider);
for t in tools {
agent.register_tool(t);
}
Some(agent)
}
struct CallMcpToolOnce {
calls: AtomicUsize,
}
#[async_trait]
impl Provider for CallMcpToolOnce {
async fn complete(
&self,
req: &ChatRequest,
_on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
) -> supercode_harness::Result<(ChatMessage, Usage)> {
let n = self.calls.fetch_add(1, Ordering::SeqCst);
if n == 0 {
let call = ChatMessage {
role: Role::Assistant,
content: None,
content_parts: None,
tool_calls: Some(vec![ToolCall {
id: "c1".into(),
kind: "function".into(),
function: FunctionCall {
name: "mcp__fake__delete_everything".into(),
arguments: "{}".into(),
},
}]),
tool_call_id: None,
name: None,
metadata: Default::default(),
};
Ok((call, Usage::default()))
} else {
let last = req.messages.last().unwrap();
Ok((
ChatMessage::assistant(last.content.clone().unwrap_or_default()),
Usage::default(),
))
}
}
}
#[tokio::test]
async fn mcp_tool_call_runs_unimpeded_when_the_permissions_engine_is_off() {
let config = Config::builder().approval(ApprovalPolicy::Never).build();
let Some(mut agent) = agent_with_mcp_tool(
config,
Box::new(CallMcpToolOnce {
calls: AtomicUsize::new(0),
}),
)
.await
else {
return;
};
let reply = agent.send("do it").await.unwrap();
assert!(reply.contains("deleted"), "reply: {reply}");
}
#[tokio::test]
async fn mcp_tool_call_is_blocked_by_a_deny_rule_matching_its_namespaced_name() {
let mut config = Config::builder().approval(ApprovalPolicy::Never).build();
config.permissions_enabled = true;
config.tool_deny_patterns = vec!["mcp__fake__*".to_string()];
let Some(mut agent) = agent_with_mcp_tool(
config,
Box::new(CallMcpToolOnce {
calls: AtomicUsize::new(0),
}),
)
.await
else {
return;
};
let reply = agent.send("do it").await.unwrap();
assert!(
reply.contains("not approved"),
"an MCP tool matching a deny rule must be blocked exactly like a native tool: {reply}"
);
}
#[tokio::test]
async fn mcp_tool_call_under_an_ask_rule_fails_closed_with_no_handler() {
let mut config = Config::builder().approval(ApprovalPolicy::Never).build();
config.permissions_enabled = true;
config.permissions_ask_patterns = vec!["mcp__fake__*".to_string()];
let Some(mut agent) = agent_with_mcp_tool(
config,
Box::new(CallMcpToolOnce {
calls: AtomicUsize::new(0),
}),
)
.await
else {
return;
};
let reply = agent.send("do it").await.unwrap();
assert!(
reply.contains("not approved"),
"an Ask-tier MCP tool call with no installed handler must fail closed: {reply}"
);
}
struct AlwaysApprove;
impl PermissionsApprovalHandler for AlwaysApprove {
fn ask(&self, _req: &ApprovalRequest) -> ApprovalOutcome {
ApprovalOutcome::Allow
}
}
#[tokio::test]
async fn mcp_tool_call_under_an_ask_rule_runs_once_a_handler_approves() {
let mut config = Config::builder().approval(ApprovalPolicy::Never).build();
config.permissions_enabled = true;
config.permissions_ask_patterns = vec!["mcp__fake__*".to_string()];
let Some(mut agent) = agent_with_mcp_tool(
config,
Box::new(CallMcpToolOnce {
calls: AtomicUsize::new(0),
}),
)
.await
else {
return;
};
agent.set_permissions_approval_handler(AlwaysApprove);
let reply = agent.send("do it").await.unwrap();
assert!(reply.contains("deleted"), "reply: {reply}");
}
#[tokio::test]
async fn mcp_tool_call_matches_an_allow_pattern_and_runs_without_a_handler() {
let mut config = Config::builder()
.approval(ApprovalPolicy::Untrusted)
.build();
config.permissions_enabled = true;
config.tool_allow_patterns = vec!["mcp__fake__*".to_string()];
let Some(mut agent) = agent_with_mcp_tool(
config,
Box::new(CallMcpToolOnce {
calls: AtomicUsize::new(0),
}),
)
.await
else {
return;
};
let reply = agent.send("do it").await.unwrap();
assert!(reply.contains("deleted"), "reply: {reply}");
}