Skip to main content

MCP, CLI, and connectors

Status: Beta Companion docs: MCP Gateway · MCP API · CLI Firewall API · Connectors Marketplace API Source of truth: MCP API, CLI Firewall API, and Connectors Marketplace API. This page shows the shipped Python calls.

Choose this when​

Choose these surfaces when an agent acts outside a model completion:

  • Use tr.mcp for tools exposed by an MCP server.
  • Use tr.cli for shell-command checks and governed execution.
  • Use tr.connectors for declarative third-party operations.

The action methods below return a standard SDK Response. Read response.ok, response.status, response.body, and response.headers.

MCP tool calls​

Register an upstream server, initialize an MCP session, then invoke the namespaced tool from the aggregated catalog.

from trinitite import Trinitite

tr = Trinitite(env="prod")

server = tr.mcp.register(
    name="github",
    url="https://mcp.example.com/v1",
    transport="streamable_http",
    auth_type="bearer",
    credential_id="cred_github_mcp",
    resync_cadence="daily",
)

catalog = tr.mcp.catalog()
tool_name = catalog.body["tools"][0]["tool_name"]

initialize = tr.client_transport.request(
    "POST",
    "/v1/mcp",
    json={
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {"name": "ops-agent", "version": "1.0.0"},
        },
    },
)
session_id = initialize.headers["mcp-session-id"]

tr.client_transport.request(
    "POST",
    "/v1/mcp",
    json={"jsonrpc": "2.0", "method": "notifications/initialized"},
    extra_headers={"Mcp-Session-Id": session_id},
)

response = tr.mcp.invoke(
    tool_name,
    arguments={"repo": "acme/service", "title": "Investigate failed job"},
    session_id=session_id,
    request_id=2,
    nhi_id="nhi_ops_agent",
)

if "error" in response.body:
    raise RuntimeError(response.body["error"])

tool_result = response.body["result"]

MCP returns a JSON-RPC envelope. A governance block appears as error code -32004. Tool names from the catalog are namespaced to avoid collisions.

CLI commands​

Use evaluate when your code will run the command. Use execute when Trinitite should evaluate and run it.

check = tr.cli.evaluate(
    "git push origin release",
    nhi_id="nhi_release_agent",
    session_id="release-2026-08-23",
    cwd="/srv/app",
    executable="/usr/bin/git",
)

verdict = check.body.get("verdict")
if verdict == "block":
    raise RuntimeError(check.body.get("violations"))

command = check.body.get("corrected_command") or "git push origin release"
run = tr.cli.execute(
    "python -m pytest",
    nhi_id="nhi_ci_agent",
    session_id="ci-4821",
)

if run.ok:
    print(run.body["execution"]["exit_code"])

CLI verdicts are pass, correct, or block. A blocked execute returns HTTP 403. A successful execute includes execution.exit_code, stdout, and stderr.

Connector operations​

There are two execution routes with different purposes.

import os

from trinitite import Config, Trinitite

# The governed connector route requires the caller's NHI token.
agent_config = Config.from_env().with_custom_header(
    "X-NHI-Token",
    os.environ["TRINITITE_NHI_TOKEN"],
)
agent_tr = Trinitite(config=agent_config)

# Admin validation only. This calls /execute and does not run agent governance.
test_response = tr.connectors.execute(
    "salesforce",
    "query",
    version="2.0.0",
    params={"soql": "SELECT Id FROM Account LIMIT 1"},
)

# Agent-facing execution. This calls /invoke and runs governance.
invoke_response = agent_tr.connectors.invoke(
    "salesforce",
    "query",
    version="2.0.0",
    params={"soql": "SELECT Id FROM Account LIMIT 1"},
    credential_set_id="ccred_salesforce_prod",
)

Use /execute only to validate a connector definition as an administrator. Use /invoke for agent traffic. The governed route applies identity and privilege checks, request and result checks, masking controls, and audit logging before it returns.

Request lifecycle​

Result and failure behavior​

  • MCP uses JSON-RPC errors inside the response body. Check response.body["error"] as well as response.ok.
  • CLI blocks return HTTP 403. Corrected evaluations can include corrected_command.
  • Connector /execute returns an execution result and is not a governed result.
  • Connector /invoke returns the agent-facing governed execution envelope. A governance block returns HTTP 403 with error.code = "connector_governance_blocked".
  • Unknown tools or connector operations return not-found responses.
  • HTTP 4xx responses are not retried. HTTP 5xx responses are retried according to SDK config. Transport errors can raise ControlPlaneError.

Identity, masking, and receipts​

Pass nhi_id to MCP and CLI calls when you need their audit rows tied to a non-human identity. Connector /invoke enforces its route's NHI identity requirements. The example adds its JIT token through Config.with_custom_header(...).

MCP and governed connector calls can apply masking before the action and to its result. Do not expect the admin connector /execute route to apply those controls.

Receipt data is surface-specific. MCP governance receipts are produced only when the active MCP governance and determinism settings emit them. CLI calls and connector calls return raw response envelopes, so inspect their documented body and audit identifiers instead of passing them to tr.result(...).

Next steps​

Last updated on