dtoro
64f7d54011
Problem: the hexagon's Phase 2 (plans/2026-08-15-hexagonal-architecture.md) must give the use-cases-to-be their contract surface: driven-port interfaces, test fakes, the secrets interface moved into core, and the postgres package inside the adapters tree — before the first vertical slice (Phase 3) can wire a composition root. Change: - internal/core/ports: full driven-port catalog per plan §3.3 — repositories as transaction-scoped aggregates whose inputs carry derived checks, audit, and events (§3.6), plus CommandExecutor, TargetResolver, Checker, Secrets, EventPublisher, Provisioner. Port-local payload types (Event, AuditEntry, CheckDef, KnowledgeEntry, ExecResult) keep signatures off infrastructure; TypeTree aliases internal/ontology (pure over domain) until checkdefaults is absorbed. ReadModels intentionally not declared yet — it materializes with the Phase 3 slice and grows as report handlers rewire. - secrets.Backend is now an alias of ports.Secrets; implementations (Infisical, SOPS, Manager) unchanged. mcp's local secretBackend subset is deleted; tool constructors take ports.Secrets. - internal/db → internal/adapters/postgres (mechanical import rewrite; package identifier stays db until the Phase 3 repository split). sqlc.yaml, Makefile, golangci exclusions, and docs follow the move; make generate-check verified. - internal/adapters/ssh: Executor implements ports.CommandExecutor over the actuator dial pool + RunStreaming (10-min default timeout carried over from the httpapi path). - internal/adapters/remote: Resolver implements ports.TargetResolver delegating to internal/remote (still pool-based; drops onto ports.EntityRepository when repositories land in Phase 3 — documented transitional import). - internal/core/ports/portstest: importable fakes — in-memory EntityRepo (with check-then-act SetState, side-effect recording), RecordingExecutor, FakeChecker, SpyPublisher; port-satisfaction guards; tests. Risk: ports are declared ahead of implementations — signatures firm up per phase as slices land (documented in the package doc); the remote→postgres transitional import is explicit and dissolves in Phase 3. Verification: go vet, make test (race, 19 packages), generate-check, golangci on core+adapters — 0 issues; full-repo baseline down 365→344.
503 lines
27 KiB
Go
503 lines
27 KiB
Go
package mcp
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/dtoro/oikos/internal/audit"
|
|
"github.com/dtoro/oikos/internal/core/ports"
|
|
"github.com/dtoro/oikos/internal/adapters/postgres"
|
|
"github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen"
|
|
"github.com/google/uuid"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
)
|
|
|
|
func EntityTools(pool *db.Pool, agentID uuid.UUID, sec ports.Secrets) []toolReg {
|
|
return []toolReg{
|
|
{tool: &mcp.Tool{Name: "ping", Description: "Lightweight connectivity check. Returns server identity, no DB hit.",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return textResult(`{"ok":true,"server":"oikos","version":"dev"}`), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_entity", Description: "Get an entity by slug or UUID",
|
|
InputSchema: objSchema(prop{"slug_or_id", "string", "Entity slug (e.g. host:hubris) or UUID"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
idOrSlug := slugArg(args, "slug_or_id", "slug", "id")
|
|
return queryEntity(ctx, pool, idOrSlug), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "list_entities", Description: "List entities filtered by type, state, or search",
|
|
InputSchema: objSchema(
|
|
prop{"type", "string", "Filter by entity type"},
|
|
prop{"state", "string", "Filter by lifecycle state"},
|
|
prop{"q", "string", "Substring match on slug or name"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state, e.version, e.created_at, e.updated_at
|
|
FROM entities e
|
|
WHERE ($1::text IS NULL OR e.type = $1)
|
|
AND ($2::text IS NULL OR e.state = $2)
|
|
AND ($3::text IS NULL OR e.slug ILIKE '%'||$3||'%' OR e.name ILIKE '%'||$3||'%')
|
|
ORDER BY e.slug LIMIT $4`,
|
|
nStr(args["type"]), nStr(args["state"]), nStr(args["q"]), limit), "entity_table"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_relations", Description: "List inbound/outbound edges for one entity, optionally filtered by relationship type",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"types", "string", "Comma-separated relationship types to include (e.g. 'hosts,provides,depends-on'). Omit for all."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug := slugArg(args, "entity_id", "slug")
|
|
typesStr, _ := args["types"].(string)
|
|
if slug == "" {
|
|
return textResult("entity_id is required"), nil
|
|
}
|
|
query := `
|
|
SELECT r.type, src.slug AS source, tgt.slug AS target
|
|
FROM relationships r
|
|
JOIN entities src ON src.id = r.source_id
|
|
JOIN entities tgt ON tgt.id = r.target_id
|
|
WHERE (src.slug = $1 OR tgt.slug = $1) AND r.valid_to IS NULL`
|
|
if typesStr != "" {
|
|
query += ` AND r.type = ANY(string_to_array($2, ','))`
|
|
return queryRows(ctx, pool, query, slug, typesStr), nil
|
|
}
|
|
query += ` ORDER BY r.type`
|
|
return queryRows(ctx, pool, query, slug), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_blast_radius", Description: "Find entities affected if this entity goes down",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"depth", "integer", "Traversal depth (default 3)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug := slugArg(args, "entity_id", "slug")
|
|
depth := int(getFloat(args, "depth", 3))
|
|
return queryRows(ctx, pool,
|
|
"SELECT e.slug, CAST(b.depth AS int) FROM blast_radius((SELECT id FROM entities WHERE slug = $1), $2) b JOIN entities e ON e.id = b.entity_id",
|
|
slug, depth), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "create_entity", Description: "Create a new entity in the knowledge graph. Use it when a task needs an entity that does not exist yet: a service, a host/LXC/VM, an ingress, a cert, etc. After inserting, it derives default checks from the entity type's monitoring spec, so creating a checkable entity wires its monitoring in one call. Does NOT require approval. If the slug already exists it returns 'already exists' — then use update_entity_attributes to change it. FOOTGUN: creating a type=check entity creates a bare entity row but does NOT wire a check_def — the scheduler will never probe it. To add monitoring, set `monitoring: [\"http\"]` + `url` on the target via update_entity_attributes.",
|
|
InputSchema: objSchema(
|
|
prop{"type", "string", "Entity type — must already exist in the ontology and not be abstract (e.g. service, lxc, host, vm, check, ingress, cert, dns)."},
|
|
prop{"name", "string", "Human-readable name (e.g. 'HAOS http service check')."},
|
|
prop{"slug", "string", "Entity slug (e.g. check:http:service:haos:0, ingress:home.hubris.network). If omitted, defaults to <type>:<name>."},
|
|
prop{"attributes", "string", "JSON object string of attributes, e.g. {\"check_type\":\"http:service\",\"target\":\"service:haos\",\"port\":\"8123\"}. Optional."},
|
|
prop{"state", "string", "Lifecycle state. Optional; defaults to the type's lifecycle default_state."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
entityType, _ := args["type"].(string)
|
|
name, _ := args["name"].(string)
|
|
slug, _ := args["slug"].(string)
|
|
if slug == "" && entityType != "" && name != "" {
|
|
slug = entityType + ":" + name
|
|
}
|
|
if entityType == "" || name == "" || slug == "" {
|
|
return textResult("error: type and name are required (slug defaults to <type>:<name>)"), nil
|
|
}
|
|
attrsStr, _ := args["attributes"].(string)
|
|
attrs := map[string]any{}
|
|
if attrsStr != "" {
|
|
if err := json.Unmarshal([]byte(attrsStr), &attrs); err != nil {
|
|
return textResult(fmt.Sprintf("error: attributes is not valid JSON: %v", err)), nil
|
|
}
|
|
}
|
|
attrsJSON, _ := json.Marshal(attrs)
|
|
stateStr, _ := args["state"].(string)
|
|
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: begin tx: %v", err)), nil
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
// Validate the type exists and is concrete (mirror httpapi.CreateEntity).
|
|
var isAbstract bool
|
|
if err := tx.QueryRow(ctx, `SELECT is_abstract FROM entity_types WHERE name = $1`, entityType).Scan(&isAbstract); err != nil {
|
|
return textResult(fmt.Sprintf("error: entity type %q not found in ontology", entityType)), nil
|
|
}
|
|
if isAbstract {
|
|
return textResult(fmt.Sprintf("error: type %q is abstract — pick a concrete subtype", entityType)), nil
|
|
}
|
|
|
|
// Default state from the type's lifecycle unless the caller
|
|
// supplied one. Caller-supplied states are validated against
|
|
// the lifecycle's declared states — a create_entity bypass of
|
|
// lifecycle guardrails would let an agent create in a terminal
|
|
// state (destroyed) without satisfying the preconditions that
|
|
// set_entity_state enforces for the same transition.
|
|
var state *string
|
|
var lsDefault, statesRaw string
|
|
if err := tx.QueryRow(ctx, `SELECT coalesce(ld.default_state,''), coalesce(ld.states::text,'')
|
|
FROM lifecycle_defs ld
|
|
JOIN entity_types et ON et.lifecycle_id = ld.id
|
|
WHERE et.name = $1`, entityType).Scan(&lsDefault, &statesRaw); err == nil {
|
|
var validStates []string
|
|
json.Unmarshal([]byte(statesRaw), &validStates)
|
|
if stateStr != "" {
|
|
found := false
|
|
for _, s := range validStates {
|
|
if s == stateStr {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found && len(validStates) > 0 {
|
|
return textResult(fmt.Sprintf("error: state %q not declared in %s lifecycle (states: %s). Use the default (%s) or omit state.", stateStr, entityType, strings.Join(validStates, ","), lsDefault)), nil
|
|
}
|
|
state = &stateStr
|
|
} else if lsDefault != "" {
|
|
state = &lsDefault
|
|
}
|
|
}
|
|
|
|
id, err := uuid.NewV7()
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: gen id: %v", err)), nil
|
|
}
|
|
|
|
var createdName string
|
|
if err := tx.QueryRow(ctx, `
|
|
INSERT INTO entities (id, slug, type, name, state, attributes)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
RETURNING name`,
|
|
id, slug, entityType, name, state, attrsJSON).Scan(&createdName); err != nil {
|
|
if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
|
|
return textResult(fmt.Sprintf("Entity %q already exists — use update_entity_attributes to change it.", slug)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("error creating %s: %v", slug, err)), nil
|
|
}
|
|
|
|
res, derr := db.EnsureEntityChecks(ctx, tx, id, slug, entityType, createdName, attrsJSON)
|
|
if derr != nil {
|
|
return textResult(fmt.Sprintf("error deriving checks for %s: %v", slug, derr)), nil
|
|
}
|
|
if cerr := tx.Commit(ctx); cerr != nil {
|
|
return textResult(fmt.Sprintf("error committing %s: %v", slug, cerr)), nil
|
|
}
|
|
|
|
return textResult(formatCreateResult(slug, entityType, res)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "update_entity_attributes", Description: "Merge new/changed attributes into an entity — the OTHER half of avoiding knowledge-base drift (upsert_knowledge records what you learned; this keeps the entity's own facts current). Use it when you discover something concrete about an entity's actual state that the graph doesn't reflect yet: a new IP, a version number, a config value, a discovered port — anything a FUTURE task would otherwise have to rediscover from scratch. Does NOT require approval (this updates the knowledge graph, not the live infrastructure). Merges shallowly — existing keys not mentioned are kept; keys you pass overwrite.",
|
|
InputSchema: objSchema(
|
|
prop{"slug", "string", "Entity slug to update (e.g. lxc:typetype, host:strong)."},
|
|
prop{"attributes", "string", "JSON object string of attributes to merge in, e.g. {\"lan_ip\":\"192.168.8.50\",\"os\":\"debian-12\"}."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["slug"].(string)
|
|
attrsStr, _ := args["attributes"].(string)
|
|
if slug == "" || attrsStr == "" {
|
|
return textResult("error: slug and attributes are required"), nil
|
|
}
|
|
var attrs map[string]any
|
|
if err := json.Unmarshal([]byte(attrsStr), &attrs); err != nil {
|
|
return textResult(fmt.Sprintf("error: attributes is not valid JSON: %v", err)), nil
|
|
}
|
|
|
|
// Strip scheduler-owned keys: health is computed by the scheduler
|
|
// from probe results (spotted live 2026-08-05: an agent set
|
|
// health:"healthy" on lxc:nfs-export, which derived 4 spurious checks).
|
|
// Agents can observe health via get_health_summary / list_checks.
|
|
var blocked []string
|
|
for _, key := range []string{"health", "last_check_at", "last_check"} {
|
|
if _, ok := attrs[key]; ok {
|
|
delete(attrs, key)
|
|
blocked = append(blocked, key)
|
|
}
|
|
}
|
|
if len(blocked) > 0 {
|
|
// Re-marshal the filtered attrs
|
|
filtered, _ := json.Marshal(attrs)
|
|
attrsStr = string(filtered)
|
|
if len(attrs) == 0 {
|
|
return textResult(fmt.Sprintf("Updated %s: no allowed attributes provided. The following keys are scheduler-owned and ignored: %s. Use get_health_summary or list_checks to observe entity health.", slug, strings.Join(blocked, ", "))), nil
|
|
}
|
|
}
|
|
attrsJSON, _ := json.Marshal(attrs)
|
|
|
|
// Run the merge + check regeneration in one transaction so the
|
|
// derived checks always see the post-merge attributes. Mirrors
|
|
// httpapi.PatchEntity; without this, setting an entity's
|
|
// `monitoring` attribute via MCP silently produced no checks.
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: begin tx: %v", err)), nil
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
ra, err := sqlcgen.New(tx).MergeEntityAttributes(ctx, sqlcgen.MergeEntityAttributesParams{Slug: slug, Patch: attrsJSON})
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error updating %s: %v", slug, err)), nil
|
|
}
|
|
if ra == 0 {
|
|
return textResult(fmt.Sprintf("error: entity %q not found", slug)), nil
|
|
}
|
|
|
|
merged, err := sqlcgen.New(tx).GetEntityBySlug(ctx, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error reloading %s: %v", slug, err)), nil
|
|
}
|
|
id := merged.ID
|
|
entityType := merged.Type
|
|
name := merged.Name
|
|
mergedAttrs := merged.Attributes
|
|
|
|
res, cerr := db.EnsureEntityChecks(ctx, tx, id, slug, entityType, name, mergedAttrs)
|
|
if cerr != nil {
|
|
return textResult(fmt.Sprintf("error deriving checks for %s: %v", slug, cerr)), nil
|
|
}
|
|
if cerr := tx.Commit(ctx); cerr != nil {
|
|
return textResult(fmt.Sprintf("error committing %s: %v", slug, cerr)), nil
|
|
}
|
|
|
|
return textResult(fmt.Sprintf("Updated %s with %d attribute(s).%s", slug, len(attrs), formatCheckResult(res))), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "set_entity_state", Description: "Transition an entity to a new lifecycle state — the entity-graph \"delete\" surface, since this system never hard-deletes entities. Use retire/deprecate to take an entity out of service, destroy for terminal removal, or active to revive. The target state must be a declared transition in the entity type's lifecycle (e.g. active→deprecated, deprecated→active); preconditions (no inbound edges, backups verified, etc.) are enforced — an error tells you what's blocking. Does NOT require approval (knowledge-graph mutation, not live infrastructure).",
|
|
InputSchema: objSchema(
|
|
prop{"slug", "string", "Entity slug."},
|
|
prop{"state", "string", "Target lifecycle state."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["slug"].(string)
|
|
targetState, _ := args["state"].(string)
|
|
if slug == "" || targetState == "" {
|
|
return textResult("error: slug and state are required"), nil
|
|
}
|
|
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: begin tx: %v", err)), nil
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
ent, err := sqlcgen.New(tx).GetEntityBySlug(ctx, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: entity %q not found", slug)), nil
|
|
}
|
|
id := ent.ID
|
|
entityType := ent.Type
|
|
currentState := ""
|
|
if ent.State != nil {
|
|
currentState = *ent.State
|
|
}
|
|
if err := db.ValidateTransition(ctx, tx, id, entityType, currentState, targetState); err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err)), nil
|
|
}
|
|
ra, err := sqlcgen.New(tx).SetEntityState(ctx, sqlcgen.SetEntityStateParams{ID: id, State: &targetState})
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error updating %s: %v", slug, err)), nil
|
|
}
|
|
if ra == 0 {
|
|
return textResult(fmt.Sprintf("error: entity %q not found", slug)), nil
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return textResult(fmt.Sprintf("error: commit: %v", err)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Transitioned %s: %s → %s.", slug, currentState, targetState)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "create_relationship", Description: "Record a relationship you discovered between two entities — the graph-structure half of keeping the knowledge base current (alongside update_entity_attributes and upsert_knowledge). Use it when you learn that one entity depends on, hosts, routes to, etc. another, and that edge isn't in the graph yet. type must be an existing relationship type (see get_relations output on similar entities for examples: hosts, provides, depends-on, configured-by, about, documents, ...). Idempotent — re-calling the same source/target/type is a no-op. Does NOT require approval.",
|
|
InputSchema: objSchema(
|
|
prop{"source", "string", "Source entity slug."},
|
|
prop{"target", "string", "Target entity slug."},
|
|
prop{"type", "string", "Relationship type name (must already exist in the ontology)."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
source, _ := args["source"].(string)
|
|
target, _ := args["target"].(string)
|
|
relType, _ := args["type"].(string)
|
|
if source == "" || target == "" || relType == "" {
|
|
return textResult("error: source, target, and type are required"), nil
|
|
}
|
|
srcEnt, err := sqlcgen.New(pool).GetEntityBySlug(ctx, source)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: source entity %q not found", source)), nil
|
|
}
|
|
tgtEnt, err := sqlcgen.New(pool).GetEntityBySlug(ctx, target)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: target entity %q not found", target)), nil
|
|
}
|
|
_, err = sqlcgen.New(pool).InsertRelationshipIfAbsent(ctx, sqlcgen.InsertRelationshipIfAbsentParams{
|
|
SourceID: srcEnt.ID, TargetID: tgtEnt.ID, Type: relType, Attributes: []byte(`{"by":"nomos"}`),
|
|
})
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error creating relationship: %v (is %q a valid relationship type?)", err, relType)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Recorded: %s —%s→ %s", source, relType, target)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "end_relationship", Description: "End an existing relationship (soft-delete by setting valid_to) — the graph-structure \"delete\" surface. Use it when you discover an edge is no longer true (a service moved hosts, a route was removed, a dependency dissolved). The edge is kept for history; only the currently-active edge is ended. Idempotent — ending an already-ended or non-existent edge is a no-op. Does NOT require approval.",
|
|
InputSchema: objSchema(
|
|
prop{"source", "string", "Source entity slug."},
|
|
prop{"target", "string", "Target entity slug."},
|
|
prop{"type", "string", "Relationship type name."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
source, _ := args["source"].(string)
|
|
target, _ := args["target"].(string)
|
|
relType, _ := args["type"].(string)
|
|
if source == "" || target == "" || relType == "" {
|
|
return textResult("error: source, target, and type are required"), nil
|
|
}
|
|
srcEnt, err := sqlcgen.New(pool).GetEntityBySlug(ctx, source)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: source entity %q not found", source)), nil
|
|
}
|
|
tgtEnt, err := sqlcgen.New(pool).GetEntityBySlug(ctx, target)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: target entity %q not found", target)), nil
|
|
}
|
|
ra, err := sqlcgen.New(pool).EndCurrentRelationship(ctx, sqlcgen.EndCurrentRelationshipParams{
|
|
SourceID: srcEnt.ID, TargetID: tgtEnt.ID, Type: relType,
|
|
})
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error ending relationship: %v", err)), nil
|
|
}
|
|
if ra == 0 {
|
|
return textResult(fmt.Sprintf("No active relationship %s —%s→ %s found.", source, relType, target)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Ended: %s —%s→ %s.", source, relType, target)), nil
|
|
}},
|
|
// ─── Client introspection tools (plan: client-lifecycle Phase 3) ──
|
|
|
|
{tool: &mcp.Tool{Name: "whoami", Description: "Get the current entity record, peers, and health for a host",
|
|
InputSchema: objSchema(prop{"hostname", "string", "Hostname of the calling machine"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hostname, _ := args["hostname"].(string)
|
|
if hostname == "" {
|
|
return textResult("error: hostname required"), nil
|
|
}
|
|
slug := "ws:" + hostname
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.attributes->>'mesh_ip' AS mesh_ip,
|
|
e.attributes->>'age_pubkey' AS age_pubkey,
|
|
e.enrolled_at
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1
|
|
ORDER BY e.slug`, slug), "entity_card"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "explain", Description: "Compact context card for a service: type, state, health, relations, risk",
|
|
InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug (e.g. service:jellyfin, lxc:caddy)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("error: service_slug required"), nil
|
|
}
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.version, e.updated_at,
|
|
COALESCE(e.attributes::text, '{}') AS attrs
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1`, slug), "entity_card"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "preflight", Description: "Risk classification for an action on a service",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Entity slug"},
|
|
prop{"action", "string", "Planned action (restart, deploy, destroy, etc.)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
action, _ := args["action"].(string)
|
|
if slug == "" || action == "" {
|
|
return textResult("error: service_slug and action required"), nil
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
CASE
|
|
WHEN $2 IN ('restart', 'logs', 'status') THEN 'reversible_low'
|
|
WHEN $2 IN ('deploy', 'upgrade', 'configure') THEN 'config_mutation'
|
|
WHEN $2 IN ('destroy', 'wipe', 'revoke') THEN 'destructive'
|
|
ELSE 'read_only'
|
|
END AS risk_class,
|
|
CASE
|
|
WHEN $2 IN ('read_only','reversible_low') THEN 'auto-act'
|
|
WHEN $2 = 'config_mutation' THEN 'operator-approval'
|
|
ELSE 'operator-approval+confirmation'
|
|
END AS approval
|
|
FROM entities e WHERE e.slug = $1`, slug, action), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_change_history", Description: "Last N change-ledger entries for an entity",
|
|
InputSchema: objSchema(
|
|
prop{"entity_slug", "string", "Entity slug"},
|
|
prop{"limit", "integer", "Max entries (default 20)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
limit := int(getFloat(args, "limit", 20))
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT al.ts AS timestamp, al.actor_type, al.actor_id::text AS actor_label,
|
|
al.action, al.method, al.path,
|
|
al.detail::text AS details, al.session_id::text AS session_id
|
|
FROM audit_log al
|
|
JOIN entities e ON e.id = al.entity_id
|
|
WHERE e.slug = $1
|
|
ORDER BY al.ts DESC
|
|
LIMIT $2`, slug, limit), "change_log"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_state_snapshot", Description: "Last scheduler Observe-pass: fleet health, disk, drift count",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.state IS NOT NULL
|
|
OR st.health IS NOT NULL
|
|
ORDER BY st.health, e.slug
|
|
LIMIT 200
|
|
`), "fleet_snapshot"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "audit_knowledge_graph", Description: "Read-only drift report over the knowledge graph and monitoring: orphan check entities, checks targeting deprecated/destroyed entities, probes stuck down/unknown, unmonitored declared entity types, and live edges pointing at destroyed targets. Returns ranked findings with a suggested remediation runbook each. Use this to validate the graph is complete and consistent before trusting health/blast-radius answers. Does NOT mutate anything.",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
findings, summary := audit.Report(ctx, pool)
|
|
b, _ := json.Marshal(map[string]any{"findings": findings, "summary": summary})
|
|
return textResult(string(b)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "discover_infra_drift", Description: "Read-only live discovery: compares running Proxmox guests (pct/qm list on every proxmox host) against the DB graph. Returns guests running with no entity (missing) and entities whose pve_id is no longer live (ghost) — drift the DB-only audit_knowledge_graph cannot see. Reaches hosts over the same SSH/pct path the checks use. Does NOT mutate anything.",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
b, _ := json.Marshal(discoverInfraDrift(ctx, pool))
|
|
return textResult(string(b)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "find_entities_by", Description: "Search entities by discovered attributes — IP address, port, version string, tag, or any key in the attributes JSONB blob. More flexible than list_entities (which filters by type/state only). Use for reverse lookups: 'what runs on port 8096?' or 'which entities have version 2.4?'",
|
|
InputSchema: objSchema(
|
|
prop{"key", "string", "Attribute key to search (e.g. ip, port, version, tag)"},
|
|
prop{"value", "string", "Value to match (case-insensitive substring)"},
|
|
prop{"limit", "integer", "Max results (default 25)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
key, _ := args["key"].(string)
|
|
val, _ := args["value"].(string)
|
|
limit := int(getFloat(args, "limit", 25))
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state, e.attributes->>$1 AS matched_value
|
|
FROM entities e
|
|
WHERE e.attributes ? $1
|
|
AND e.attributes->>$1 ILIKE '%'||$2||'%'
|
|
ORDER BY e.slug
|
|
LIMIT $3`, key, val, limit), nil
|
|
}},
|
|
}
|
|
}
|