MCP Server Specifications — Project VELA
Three servers. FastMCP (Python). Agents access data ONLY through these tools — this is the audit boundary.
Design rules (apply to every tool): typed Pydantic I/O; every response includes provenance (query + store + timestamp) so citations are mechanical; read-only — no tool anywhere may construct or send telecommands; errors return structured {error, hint} (never empty strings) so agents can self-correct; paginate anything unbounded.
1. vela-telemetry-mcp
Section titled “1. vela-telemetry-mcp”| Tool | Input → Output | Notes |
|---|---|---|
list_channels |
{satellite, subsystem?} → channel metadata (units, limits, criticality) |
metadata mirrored from KG |
get_window |
{satellite, channels[], t_start, t_end, max_points?} → samples + stats (min/max/mean/σ, limit crossings) |
server-side decimation; agents never pull raw megapoints |
detect_in_window |
{satellite, channels[], t_start, t_end, detector} → point scores + formed events |
wraps Layer-1; lets validator re-check a hypothesis window |
get_active_events |
{satellite} → open AnomalyEvents |
|
get_command_history |
{satellite, t_start, t_end, related_to_channel?} → TelecommandRecord[] |
related_to_channel resolves via KG AFFECTS |
compare_to_baseline |
{satellite, channel, window, baseline: prior_orbit|prior_day|training_dist} → deltas |
powers “is this actually unusual” checks |
2. vela-knowledge-mcp
Section titled “2. vela-knowledge-mcp”| Tool | Input → Output | Notes |
|---|---|---|
get_component_tree |
{satellite, subsystem?} → subsystem→component tree |
|
channels_to_components |
{channels[]} → components measured, with relation paths |
first hop of every investigation |
get_failure_modes |
{component_id} → failure modes + expected telemetry signatures + severity |
signature = the testable prediction |
traverse |
{start_id, relation_whitelist[], max_hops≤3} → subgraph |
whitelist + hop cap = no free-form Cypher from LLMs |
search_docs |
{query, subsystem?, k≤8} → chunks with {doc_id, section_ref, page, sha256} |
Qdrant hybrid search |
get_doc_section |
{doc_id, section_ref} → full section text |
for exact citation quoting internally |
find_similar_cases |
{event_summary, channels[], k≤5} → past AnomalyCases + resolutions |
case-based reasoning memory |
resolve_citation |
{Citation} → {valid: bool, target_excerpt} |
used by the audit checker, exposed for agents to self-verify |
3. vela-procedure-mcp
Section titled “3. vela-procedure-mcp”| Tool | Input → Output | Notes |
|---|---|---|
get_procedure_templates |
{failure_mode_id?} → procedure skeletons from MITIGATED_BY + ECSS-style structure |
|
validate_procedure |
{ProcStep[]} → per-step {grounded: bool, citations, risk_flags} |
risk flags: irreversible action, thermal/power precondition missing, etc. (rule-based v1) |
format_report |
{InvestigationReport} → operator-facing markdown/PDF |
rendering only; no new content generation |
4. Existing external MCP servers — verify before building duplicates
Section titled “4. Existing external MCP servers — verify before building duplicates”Community MCP servers already exist for adjacent public space data (NASA open APIs, TLE/orbit data via CelesTrak/Space-Track wrappers). Before writing any auxiliary integration, search your own WebMCP Registry and scan candidates with MCPScan.dev — dogfooding both products is a story worth telling Dhruva. Rule: external MCP servers may inform context (orbital events, space weather) but are never citation sources for spacecraft-specific claims — only the customer’s own docs in the KG are.
5. Skeleton (pattern for all three servers)
Section titled “5. Skeleton (pattern for all three servers)”from fastmcp import FastMCPfrom vela.contracts import AnomalyEvent, Provenancemcp = FastMCP("vela-telemetry")
@mcp.tool()def get_window(satellite: str, channels: list[str], t_start: str, t_end: str, max_points: int = 2000) -> dict: """Fetch decimated telemetry with summary stats. Never fabricate: raises if store lacks range.""" data = store.window(satellite, channels, t_start, t_end, max_points) return {"samples": data.samples, "stats": data.stats, "provenance": Provenance(query=locals(), store=store.id).model_dump()}Testing per server: pytest against docker-compose fixtures; golden request/response snapshots; a “hostile agent” test that calls tools with malformed args and asserts structured errors. CI runs all three in cloud and airgap config.
