v2.0.1

MCP Controlled Writes

This tutorial demonstrates controlled write operations through the MCP server: submitting evaluations, creating experiments, managing jobs, and looking up audit records.

Overview

Controlled write tools share a required envelope with request ID, idempotency key, actor metadata, and optional dry-run mode. Every write is recorded in the MCP audit log with full provenance.

Write envelope

All controlled writes require this envelope:

json

{
  "request_id": "request-1",
  "correlation_id": "corr-1",
  "idempotency_key": "idem-key-1",
  "requested_by": "agent-live",
  "actor_type": "AGENT",
  "reason": "Submit evaluation for claim validation",
  "metadata": {"ticket": "KAV-123"}
}

Submit evaluation via MCP

Submit an async evaluation through the MCP server:

python

from kavach import create_mcp_server

server = create_mcp_server()

result = server.call_tool(
    "evaluation.submit_async",
    {
        "request_id": "request-write-evaluation",
        "correlation_id": "corr-write",
        "idempotency_key": "idem-write-evaluation",
        "requested_by": "agent-live",
        "actor_type": "AGENT",
        "reason": "Submit evaluation for claim validation",
        "provider_name": "mock",
        "workflow_id": "claim-validation",
        "workflow_name": "Claim Validation",
        "execution_id": "execution-write-1",
        "input": {"question": "Was the claim valid?"},
        "final_state": {"answer": "The claim appears valid."},
        "metric_specs": [{"name": "answer_relevance"}]
    }
)

print(f"job_id={result.data['job_id']}")

Create experiment via MCP

Create an experiment through the MCP server:

python

result = server.call_tool(
    "experiment.create",
    {
        "request_id": "request-write-experiment",
        "correlation_id": "corr-write",
        "idempotency_key": "idem-write-experiment",
        "requested_by": "agent-live",
        "actor_type": "AGENT",
        "reason": "Create experiment for smoke test",
        "name": "smoke-experiment",
        "description": "Smoke-test controlled experiment creation."
    }
)

print(f"job_id={result.data['job_id']}")

Job management

Check job status and cancel jobs through MCP:

python

# Check job status
status = server.call_tool(
    "job.status",
    {"job_id": evaluation_job_id}
)

# Cancel a job with controlled write envelope
cancelled = server.call_tool(
    "job.cancel",
    {
        "request_id": "request-write-cancel",
        "correlation_id": "corr-write",
        "idempotency_key": "idem-write-cancel",
        "requested_by": "agent-live",
        "actor_type": "AGENT",
        "reason": "Cancel evaluation job",
        "job_id": evaluation_job_id
    }
)

print(f"cancelled={cancelled.data['status']}")

Audit lookup

Find audit records by correlation ID:

python

audit = server.call_tool(
    "mcp_audit.find_by_correlation",
    {"correlation_id": "corr-write"}
)

records = audit.data["records"]
for record in records:
    print(f"tool={record['tool_name']} status={record['status']}")

Dry run mode

When dry_run is true, MCP validates the envelope and tool-specific schema, records the audit entry, and returns a validation result without calling REST or creating a job.

python

result = server.call_tool(
    "evaluation.submit_async",
    {
        "request_id": "request-dry-run",
        "correlation_id": "corr-dry",
        "idempotency_key": "idem-dry-run",
        "requested_by": "agent-live",
        "actor_type": "AGENT",
        "reason": "Dry-run evaluation submission",
        "dry_run": True,  # <-- validates without executing
        "provider_name": "mock",
        "workflow_id": "claim-validation",
        "execution_id": "exec-dry-run",
        "metric_specs": [{"name": "answer_relevance"}]
    }
)

print(f"status={result.status}")  # "ok" if validation passes