646 lines
22 KiB
Python
646 lines
22 KiB
Python
"""Native Artemis MCP entry points for the user-provided multi-tool testing Skill."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from functools import wraps
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import tempfile
|
||
from typing import Annotated, Literal
|
||
|
||
from mcp.server.fastmcp import FastMCP
|
||
from mcp.types import CallToolResult, ToolAnnotations
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
from device_runtime import (
|
||
InputError,
|
||
PACKAGE,
|
||
artifact_root,
|
||
attach_file,
|
||
check_budget,
|
||
failure,
|
||
local_file,
|
||
new_artifact,
|
||
receipt,
|
||
rtsp_environment,
|
||
run_cli,
|
||
runner_arguments,
|
||
sanitized,
|
||
sha256,
|
||
task_evidence,
|
||
unicast_host,
|
||
)
|
||
|
||
|
||
mcp = FastMCP("artemis-device-test")
|
||
Text = Annotated[str, Field(min_length=1, max_length=4096)]
|
||
Identity = Annotated[str, Field(min_length=1, max_length=256)]
|
||
Protocol = Literal["v1", "v2"]
|
||
VersionField = Literal["firmware", "dev_ver", "inner.version", "outer.version"]
|
||
Seconds = Annotated[int, Field(ge=1, le=300)]
|
||
|
||
|
||
def tool(*, write: bool = False):
|
||
def register(function):
|
||
@wraps(function)
|
||
async def guarded(*args, **kwargs):
|
||
try:
|
||
return await function(*args, **kwargs)
|
||
except InputError as exc:
|
||
return failure(str(exc))
|
||
except (OSError, ValueError, TypeError):
|
||
return failure(
|
||
"本机文件或配置无法读取；未将异常原文写入回执。",
|
||
"local_configuration_error",
|
||
execution="unknown",
|
||
side_effect="possible" if write else "none",
|
||
)
|
||
|
||
mcp.tool(
|
||
annotations=ToolAnnotations(
|
||
readOnlyHint=not write,
|
||
destructiveHint=write,
|
||
idempotentHint=not write,
|
||
openWorldHint=True,
|
||
)
|
||
)(guarded)
|
||
return guarded
|
||
|
||
return register
|
||
|
||
|
||
async def runner(
|
||
command: str,
|
||
arguments: list[str],
|
||
*,
|
||
target="local-environment",
|
||
files=None,
|
||
env=None,
|
||
secrets=(),
|
||
write=False,
|
||
):
|
||
return await run_cli(
|
||
"dependency_runner.py",
|
||
runner_arguments(command, arguments),
|
||
operation=command,
|
||
target=target,
|
||
files=files,
|
||
env=env,
|
||
secrets=secrets,
|
||
write=write,
|
||
)
|
||
|
||
|
||
def identity(
|
||
protocol: Protocol, host: str, expected_sn: str, expected_model: str | None, timeout: int
|
||
) -> list[str]:
|
||
if not expected_sn.strip() or (expected_model is not None and not expected_model.strip()):
|
||
raise InputError("必须提供完整 SN 和设备上报的准确型号。")
|
||
check_budget(timeout + 5)
|
||
arguments = [
|
||
"--protocol",
|
||
protocol,
|
||
"--host",
|
||
unicast_host(host),
|
||
"--expected-sn",
|
||
expected_sn,
|
||
"--timeout",
|
||
str(timeout),
|
||
]
|
||
if expected_model is not None:
|
||
arguments += ["--expected-model", expected_model]
|
||
return arguments
|
||
|
||
|
||
def version(protocol: Protocol, target_version: str, version_field: VersionField) -> list[str]:
|
||
allowed = {"firmware", "dev_ver"} if protocol == "v1" else {"inner.version", "outer.version"}
|
||
if version_field not in allowed or not target_version.strip():
|
||
raise InputError(
|
||
"验收字段与设备协议不符；V1 使用 firmware/dev_ver，V2 使用 inner.version/outer.version。"
|
||
)
|
||
return ["--target-version", target_version, "--version-field", version_field]
|
||
|
||
|
||
@tool()
|
||
async def environment_doctor(
|
||
check_versions: bool = False, check_functional: bool = False
|
||
) -> CallToolResult:
|
||
"""逐项检查 PATH/统一目录/profile。check_versions 验证启动/DLL；check_functional 加做固定本地样本，不扫描、抓包或打开串口。缺失只阻塞相关能力。"""
|
||
arguments = ["--check-versions"] if check_versions else []
|
||
if check_functional:
|
||
arguments.append("--check-functional")
|
||
return await runner("doctor", arguments)
|
||
|
||
|
||
@tool()
|
||
async def lookup_case(
|
||
sheet: Annotated[str, Field(max_length=256)] = "",
|
||
row: Annotated[int, Field(ge=0, le=1000000)] = 0,
|
||
query: Annotated[str, Field(max_length=256)] = "",
|
||
limit: Annotated[int, Field(ge=1, le=20)] = 10,
|
||
) -> CallToolResult:
|
||
"""查询原 Skill 的 357 条路由；用 sheet+row 唯一定位，query 可查编号/标题；路由不是完整用例。"""
|
||
if row and not sheet:
|
||
raise InputError("按行查询必须同时提供工作表 sheet。")
|
||
catalog = json.loads((PACKAGE / "skill" / "assets" / "case-routing.json").read_text("utf-8"))
|
||
matches = []
|
||
for case in catalog["cases"]:
|
||
source = case["source"]
|
||
if sheet and source["sheet"] != sheet:
|
||
continue
|
||
if row and source["row"] != row:
|
||
continue
|
||
if query and query.casefold() not in json.dumps(case, ensure_ascii=False).casefold():
|
||
continue
|
||
matches.append(case)
|
||
return receipt(
|
||
"case-lookup",
|
||
{
|
||
"total_matches": len(matches),
|
||
"cases": matches[:limit],
|
||
"complete_case_specification_required": True,
|
||
},
|
||
kind="file",
|
||
target="skill/assets/case-routing.json",
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def probe_media(
|
||
input_file: Annotated[str, Field(max_length=4096)] = "",
|
||
rtsp_env: Annotated[str, Field(max_length=100)] = "",
|
||
transport: Literal["tcp", "udp"] = "tcp",
|
||
) -> CallToolResult:
|
||
"""ffprobe 白名单媒体元数据；input_file 与 rtsp_env 二选一。RTSP 只收本机环境变量名。"""
|
||
if bool(input_file) == bool(rtsp_env):
|
||
raise InputError("本地文件和 RTSP 环境变量必须且只能提供一个。")
|
||
env, secrets = rtsp_environment(rtsp_env) if rtsp_env else ({}, ())
|
||
arguments = (
|
||
["--rtsp-env", rtsp_env, "--transport", transport]
|
||
if rtsp_env
|
||
else ["--input", str(local_file(input_file))]
|
||
)
|
||
return await runner(
|
||
"probe",
|
||
arguments,
|
||
target="rtsp-env:" + rtsp_env if rtsp_env else input_file,
|
||
env=env,
|
||
secrets=secrets,
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def media_info(input_file: Text) -> CallToolResult:
|
||
"""通过 profile 指定的 MediaInfo.dll 有界分析本地媒体，输出固定元数据；不调用 GUI。"""
|
||
return await runner("media-info", ["--input", str(local_file(input_file))], target=input_file)
|
||
|
||
|
||
@tool()
|
||
async def analyze_pcap(
|
||
input_file: Text, max_packets: Annotated[int, Field(ge=1, le=10000)] = 1000
|
||
) -> CallToolResult:
|
||
"""TShark 离线提取最多 10000 包的固定头字段，不返回 payload 或认证头。"""
|
||
return await runner(
|
||
"analyze-pcap",
|
||
["--input", str(local_file(input_file)), "--max-packets", str(max_packets)],
|
||
target=input_file,
|
||
)
|
||
|
||
|
||
@tool(write=True)
|
||
async def capture_packets(
|
||
interface: Identity, host: Identity, duration_seconds: Seconds = 10
|
||
) -> CallToolResult:
|
||
"""显式网卡+单 IP 的有界抓包；仅在当前用例授权采集时使用，输出到本任务目录。"""
|
||
host = unicast_host(host)
|
||
check_budget(duration_seconds + 8)
|
||
output = new_artifact("capture", ".pcapng")
|
||
return await runner(
|
||
"capture",
|
||
[
|
||
"--interface",
|
||
interface,
|
||
"--host",
|
||
host,
|
||
"--duration",
|
||
str(duration_seconds),
|
||
"--output",
|
||
str(output),
|
||
],
|
||
target=host,
|
||
files=[output],
|
||
write=True,
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def scan_ports(
|
||
host: Identity,
|
||
ports: Annotated[
|
||
list[Annotated[int, Field(ge=1, le=65535)]], Field(min_length=1, max_length=128)
|
||
],
|
||
) -> CallToolResult:
|
||
"""Nmap 单 IP 的最多 128 个明确 TCP 端口检查；无网段、NSE、服务版本探测。"""
|
||
host = unicast_host(host)
|
||
return await runner(
|
||
"scan", ["--host", host, "--ports", ",".join(map(str, sorted(set(ports))))], target=host
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def scan_firmware(input_file: Text) -> CallToolResult:
|
||
"""Binwalk 本地固件签名扫描；不解包、不运行提取物，解包失败不能证明 Flash 已加密。"""
|
||
return await runner(
|
||
"firmware-scan", ["--input", str(local_file(input_file))], target=input_file
|
||
)
|
||
|
||
|
||
@tool(write=True)
|
||
async def capture_serial(
|
||
port: Identity,
|
||
baud_rate: Annotated[int, Field(ge=1, le=4000000)],
|
||
duration_seconds: Seconds = 10,
|
||
dtr_enable: bool = False,
|
||
rts_enable: bool = False,
|
||
) -> CallToolResult:
|
||
"""明确 COM/波特率的有界采样，不发送串口命令；DTR/RTS 默认关闭，开启可能改变设备状态。"""
|
||
if not re.fullmatch(r"COM[1-9][0-9]{0,3}", port, re.IGNORECASE):
|
||
raise InputError("串口必须是明确的 COM 口。")
|
||
check_budget(duration_seconds + 8)
|
||
output = new_artifact("serial", ".txt")
|
||
arguments = [
|
||
"--port",
|
||
port.upper(),
|
||
"--baud",
|
||
str(baud_rate),
|
||
"--duration",
|
||
str(duration_seconds),
|
||
"--output",
|
||
str(output),
|
||
]
|
||
if dtr_enable:
|
||
arguments.append("--dtr-enable")
|
||
if rts_enable:
|
||
arguments.append("--rts-enable")
|
||
return await runner(
|
||
"serial",
|
||
arguments,
|
||
target=port.upper(),
|
||
files=[output, Path(str(output) + ".raw.bin")],
|
||
write=True,
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def device_capabilities() -> CallToolResult:
|
||
"""离线读取设备适配器已实现的 V1/V2 操作与字段；不接触设备，不代表真机验证。"""
|
||
return await run_cli(
|
||
"device_adapter.py", ["capabilities"], operation="device-capabilities", kind="api"
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def list_network_interfaces() -> CallToolResult:
|
||
"""只读枚举已连接的本机 IPv4 网卡，返回可用于发现的 IPv4/PREFIX；不发送设备发现包。"""
|
||
return await run_cli(
|
||
"network_inventory.py", [], operation="network-interfaces", kind="computer"
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def discover_devices(
|
||
protocol: Protocol,
|
||
interfaces: Annotated[list[Identity], Field(min_length=1, max_length=8)],
|
||
expected_sn: Annotated[str, Field(max_length=256)] = "",
|
||
expected_model: Annotated[str, Field(max_length=256)] = "",
|
||
wait_seconds: Annotated[int, Field(ge=1, le=10)] = 3,
|
||
) -> CallToolResult:
|
||
"""在本机 IPv4/PREFIX 上只读发现。先 list_network_interfaces；未知设备盘点可分别查 v1/v2。"""
|
||
check_budget(wait_seconds * len(interfaces) + 5)
|
||
arguments = ["discover", "--protocol", protocol, "--timeout", str(wait_seconds)]
|
||
for interface in interfaces:
|
||
try:
|
||
value = ipaddress.IPv4Interface(interface)
|
||
if value.ip.is_multicast or value.ip.is_unspecified or value.network.prefixlen > 30:
|
||
raise ValueError
|
||
except ValueError:
|
||
raise InputError("发现网卡须为明确的本机 IPv4/PREFIX，前缀不大于 30。") from None
|
||
arguments += ["--interface", str(value)]
|
||
if expected_sn:
|
||
arguments += ["--expected-sn", expected_sn]
|
||
if expected_model:
|
||
arguments += ["--expected-model", expected_model]
|
||
return await run_cli(
|
||
"device_adapter.py",
|
||
arguments,
|
||
operation="device-discover",
|
||
target=",".join(interfaces),
|
||
kind="api",
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def identify_device(
|
||
protocol: Protocol,
|
||
host: Identity,
|
||
expected_sn: Identity,
|
||
timeout_seconds: Annotated[int, Field(ge=1, le=20)] = 5,
|
||
) -> CallToolResult:
|
||
"""已知完整 SN 时只读观察实际上报型号；型号未核验前不能升级。"""
|
||
arguments = identity(protocol, host, expected_sn, None, timeout_seconds)
|
||
return await run_cli(
|
||
"device_adapter.py",
|
||
["identify", *arguments],
|
||
operation="device-identify",
|
||
target=unicast_host(host),
|
||
kind="api",
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def device_info(
|
||
protocol: Protocol,
|
||
host: Identity,
|
||
expected_sn: Identity,
|
||
expected_model: Identity,
|
||
timeout_seconds: Annotated[int, Field(ge=1, le=20)] = 5,
|
||
) -> CallToolResult:
|
||
"""按明确协议只读查询，并精确核验完整 SN 与上报型号。"""
|
||
arguments = identity(protocol, host, expected_sn, expected_model, timeout_seconds)
|
||
return await run_cli(
|
||
"device_adapter.py",
|
||
["info", *arguments],
|
||
operation="device-info",
|
||
target=unicast_host(host),
|
||
kind="api",
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def inspect_firmware(
|
||
rom_file: Text, expected_sha256: Annotated[str, Field(max_length=64)] = ""
|
||
) -> CallToolResult:
|
||
"""离线检查本地 ROM 格式和 SHA-256；不联网。"""
|
||
arguments = ["offline", "--rom", str(local_file(rom_file))]
|
||
if expected_sha256:
|
||
if not re.fullmatch(r"[a-fA-F0-9]{64}", expected_sha256):
|
||
raise InputError("ROM 摘要必须是完整的 SHA-256。")
|
||
arguments += ["--rom-sha256", expected_sha256]
|
||
return await run_cli(
|
||
"device_adapter.py", arguments, operation="firmware-inspect", target=rom_file, kind="file"
|
||
)
|
||
|
||
|
||
async def firmware_operation(
|
||
protocol,
|
||
host,
|
||
expected_sn,
|
||
expected_model,
|
||
rom_file,
|
||
rom_sha256,
|
||
target_version,
|
||
version_field,
|
||
execute=False,
|
||
review_summary="",
|
||
transfer_timeout_seconds=30,
|
||
):
|
||
if not re.fullmatch(r"[a-fA-F0-9]{64}", rom_sha256):
|
||
raise InputError("固件流程需要完整的已审阅 ROM SHA-256。")
|
||
arguments = [
|
||
"upgrade",
|
||
*identity(protocol, host, expected_sn, expected_model, 5),
|
||
*version(protocol, target_version, version_field),
|
||
"--rom",
|
||
str(local_file(rom_file)),
|
||
"--rom-sha256",
|
||
rom_sha256.lower(),
|
||
]
|
||
if execute:
|
||
if os.environ.get("ARTEMIS_DEVICE_TEST_ALLOW_FIRMWARE_WRITE", "false").casefold() != "true":
|
||
raise InputError(
|
||
"固件写入默认关闭；当前调用未传包。需在本机插件设置明确允许后，按当前用例授权执行。"
|
||
)
|
||
if len(review_summary.strip()) < 20:
|
||
raise InputError(
|
||
"执行升级须提供审阅摘要，说明当前授权、目标身份、ROM 兼容性与指定验收字段。"
|
||
)
|
||
check_budget(transfer_timeout_seconds + 10)
|
||
arguments += [
|
||
"--execute",
|
||
"--stall-timeout",
|
||
str(min(15, transfer_timeout_seconds)),
|
||
"--total-timeout",
|
||
str(transfer_timeout_seconds),
|
||
]
|
||
return await run_cli(
|
||
"device_adapter.py",
|
||
arguments,
|
||
operation="firmware-upgrade" if execute else "firmware-preflight",
|
||
target=unicast_host(host),
|
||
kind="api",
|
||
write=execute,
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def firmware_preflight(
|
||
protocol: Protocol,
|
||
host: Identity,
|
||
expected_sn: Identity,
|
||
expected_model: Identity,
|
||
rom_file: Text,
|
||
rom_sha256: Identity,
|
||
target_version: Identity,
|
||
version_field: VersionField,
|
||
) -> CallToolResult:
|
||
"""联网只读预检身份及本地 ROM，不传包；不以配置软件版本推断设备协议。"""
|
||
return await firmware_operation(
|
||
protocol,
|
||
host,
|
||
expected_sn,
|
||
expected_model,
|
||
rom_file,
|
||
rom_sha256,
|
||
target_version,
|
||
version_field,
|
||
)
|
||
|
||
|
||
@tool(write=True)
|
||
async def upgrade_firmware(
|
||
protocol: Protocol,
|
||
host: Identity,
|
||
expected_sn: Identity,
|
||
expected_model: Identity,
|
||
rom_file: Text,
|
||
rom_sha256: Identity,
|
||
target_version: Identity,
|
||
version_field: VersionField,
|
||
execute: bool = False,
|
||
review_summary: Annotated[str, Field(max_length=4000)] = "",
|
||
transfer_timeout_seconds: Annotated[int, Field(ge=1, le=3000)] = 30,
|
||
) -> CallToolResult:
|
||
"""默认仅预检；本机允许写入+execute=true+审阅摘要才最多传包一次。成功后仍须同 SN 版本回读，不重试。"""
|
||
return await firmware_operation(
|
||
protocol,
|
||
host,
|
||
expected_sn,
|
||
expected_model,
|
||
rom_file,
|
||
rom_sha256,
|
||
target_version,
|
||
version_field,
|
||
execute,
|
||
review_summary,
|
||
transfer_timeout_seconds,
|
||
)
|
||
|
||
|
||
@tool()
|
||
async def verify_firmware(
|
||
protocol: Protocol,
|
||
host: Identity,
|
||
expected_sn: Identity,
|
||
expected_model: Identity,
|
||
target_version: Identity,
|
||
version_field: VersionField,
|
||
timeout_seconds: Annotated[int, Field(ge=1, le=20)] = 5,
|
||
) -> CallToolResult:
|
||
"""升级后独立只读回查同 SN/型号及原用例指定固件字段；仅核验该字段，不生成整条用例结论。"""
|
||
arguments = [
|
||
"verify-version",
|
||
*identity(protocol, host, expected_sn, expected_model, timeout_seconds),
|
||
*version(protocol, target_version, version_field),
|
||
]
|
||
return await run_cli(
|
||
"device_adapter.py",
|
||
arguments,
|
||
operation="firmware-verify",
|
||
target=unicast_host(host),
|
||
kind="api",
|
||
)
|
||
|
||
|
||
@tool(write=True)
|
||
async def import_evidence(
|
||
input_file: Text, expected_sha256: Annotated[str, Field(max_length=64)] = ""
|
||
) -> CallToolResult:
|
||
"""将当前用例明确选择的本机证据复制到任务资料；最多 64 MiB，返回回读 SHA-256，不执行文件。"""
|
||
path = local_file(input_file)
|
||
if path.stat().st_size > 64 * 1024 * 1024:
|
||
raise InputError("单份证据不能超过 64 MiB。")
|
||
digest = sha256(path)
|
||
if expected_sha256 and expected_sha256.lower() != digest:
|
||
raise InputError("输入证据与期望 SHA-256 不匹配；未归档。")
|
||
attachment = attach_file(path)
|
||
if attachment["sha256"] != digest:
|
||
raise InputError("证据在导入期间发生变化；未确认归档完整性。")
|
||
return receipt(
|
||
"evidence-import",
|
||
{"imported_evidence": attachment},
|
||
target=input_file,
|
||
kind="file",
|
||
side_effect="confirmed",
|
||
files=[artifact_root() / attachment["name"]],
|
||
)
|
||
|
||
|
||
class EvidenceRef(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
path: Identity
|
||
sha256: Annotated[str | None, Field(pattern=r"^[a-fA-F0-9]{64}$")] = None
|
||
|
||
|
||
class Assertion(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
id: Identity
|
||
required: bool = True
|
||
expected: Text
|
||
actual: Text
|
||
verdict: Literal["passed", "failed", "blocked", "inconclusive"]
|
||
evidence: Annotated[list[Identity | EvidenceRef], Field(max_length=128)] = Field(
|
||
default_factory=list
|
||
)
|
||
|
||
|
||
class CaseContext(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
sheet: Identity
|
||
row: Annotated[int, Field(ge=1, le=1000000)]
|
||
case_id: Identity
|
||
|
||
|
||
class ReportContext(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
execution_mode: Literal["live", "offline", "simulation"]
|
||
real_device_test_executed: bool = False
|
||
case: CaseContext | None = None
|
||
artemis_session_id: Annotated[str, Field(max_length=256)] = ""
|
||
notes: Annotated[str, Field(max_length=4000)] = ""
|
||
manual_verification: Annotated[list[Text], Field(max_length=50)] = Field(default_factory=list)
|
||
|
||
|
||
@tool(write=True)
|
||
async def bundle_evidence(
|
||
run_id: Identity,
|
||
assertions: Annotated[list[Assertion], Field(max_length=128)],
|
||
context: ReportContext,
|
||
) -> CallToolResult:
|
||
"""依据显式观测断言生成独立 Markdown/JSON/ZIP；只引用本任务附件，缺失或哈希不符的 PASS 降为未验证。"""
|
||
if not re.fullmatch(r"[A-Za-z0-9_.-]{1,100}", run_id):
|
||
raise InputError("run_id 仅允许 1..100 个字母、数字、下划线、点或连字符。")
|
||
if context.execution_mode != "live" and context.real_device_test_executed:
|
||
raise InputError("离线/模拟记录不能声明已执行真机测试。")
|
||
files = set()
|
||
normalized = []
|
||
for assertion in assertions:
|
||
item = assertion.model_dump(exclude_none=True)
|
||
for ref in item["evidence"]:
|
||
name = ref if isinstance(ref, str) else ref["path"]
|
||
file = task_evidence(name, allow_missing=True)
|
||
if file.is_file():
|
||
files.add(file)
|
||
if isinstance(ref, dict) and ref.get("sha256"):
|
||
ref["sha256"] = ref["sha256"].lower()
|
||
normalized.append(item)
|
||
if sum(path.stat().st_size for path in files) > 48 * 1024 * 1024:
|
||
raise InputError(
|
||
"联合包输入证据总计不能超过 48 MiB；请为大文件保留单独附件并按需选取证据。"
|
||
)
|
||
# The imported bundler executes in a cancellable worker; staging never enters task exports.
|
||
with tempfile.TemporaryDirectory(prefix="artemis-device-evidence-") as staging:
|
||
directory = Path(staging)
|
||
manifest = directory / "input.json"
|
||
manifest.write_text(
|
||
json.dumps(
|
||
sanitized(
|
||
{
|
||
"schema_version": 1,
|
||
"run_id": run_id,
|
||
"evidence_root": str(artifact_root()),
|
||
"context": context.model_dump(exclude_none=True),
|
||
"assertions": normalized,
|
||
}
|
||
),
|
||
ensure_ascii=False,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
output = directory / "result"
|
||
return await run_cli(
|
||
"evidence.py",
|
||
["bundle", "--manifest", str(manifest), "--out", str(output)],
|
||
operation="evidence-bundle",
|
||
kind="file",
|
||
target=run_id,
|
||
write=True,
|
||
files=[output / "report.md", output / "manifest.json", output / "bundle.zip"],
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
mcp.run(transport="stdio")
|