cursus.mcp

cursus.mcp — Cursus as an agentic toolset.

This module exposes the cursus pipeline-engine capabilities as framework-neutral tools: small, typed functions that take JSON-serializable arguments and return a JSON-serializable ToolResult. The same tool functions are consumable by any agent framework (MCP, OpenAI function-calling, Claude tool_use, Bedrock), and a thin optional adapter (cursus.mcp.server) mounts them on an actual MCP server.

Design (parallel to core / steps — it does not replace the human-facing cursus.cli):

• envelope — the ToolResult success/error envelope every tool returns.

• tools/* — one file per namespace (catalog, dag, config, compile, validate,

  execdoc, pipeline_catalog), each a set of pure tool functions.

• schemas — not a module; each tool carries its JSON schema via ToolDef.

• registry — the canonical name → ToolDef mapping (catalog.list_steps …).

• server — optional MCP server adapter (imports the mcp SDK lazily).

Quick start:

from cursus.mcp import get_registry, call_tool

reg = get_registry()                       # name -> ToolDef
reg["catalog.list_steps"].schema           # JSON schema for the tool
result = call_tool("catalog.list_steps", {})   # or {"job_type": "training"}
result.ok                                  # True / False
result.data                                # JSON-serializable payload

The registry is the single source of truth; the MCP server, an OpenAI tool list, and the CLI (in future) can all be generated from it.

class ToolResult(ok, data=None, error=None, code=None, warnings=<factory>, next_steps=<factory>, remedy=None, meta=<factory>)

Bases: object

Uniform tool outcome.

ok

Whether the tool succeeded.

Type:

bool

data

JSON-serializable payload on success (None on error).

Type:

Any | None

error

Human-readable error message on failure (None on success).

Type:

str | None

code

Machine-readable error code on failure (e.g. "not_found", "invalid_input", "internal_error").

Type:

str | None

warnings

Non-fatal messages an agent may surface or reason about.

Type:

List[str]

next_steps

Optional in-band guidance for an agent on a successful result — a list of {"tool", "when", "why"} ("args_hint" optional) entries naming the tool(s) the agent would typically call next. Lets the golden path live on the result instead of in the agent’s prompt.

Type:

List[Dict[str, Any]]

remedy

Optional in-band recovery hint on a failure — {"suggested_tools": [...], "fix_action": str} naming the tool(s) or action that typically resolves this error, so the agent does not have to reverse-engineer a remedy from the bare code.

Type:

Dict[str, Any] | None

meta

Optional side-band info (tool name, timings, counts) — never required for correctness.

Type:

Dict[str, Any]

code: str | None = None

data: Any | None = None

error: str | None = None

classmethod failure(message, code='tool_error', details=None, warnings=None, remedy=None)

remedy: Dict[str, Any] | None = None

classmethod success(data=None, warnings=None, next_steps=None, **meta)

to_dict()

Render to a plain JSON-serializable dict (drops empty optional fields).

ok: bool

warnings: List[str]

next_steps: List[Dict[str, Any]]

meta: Dict[str, Any]

exception ToolError(message, code='tool_error', details=None)

Bases: Exception

Raised inside a tool to signal a handled, user-facing failure.

The registry’s invoker catches this and converts it to ToolResult.failure(...), preserving code and details. Use this for expected failures (bad input, not-found, validation) — let genuinely unexpected exceptions propagate so the invoker can wrap them as an internal_error.

class ToolDef(name, description, schema, handler, destructive=False, tags=(), when='', examples=(), writes=False, exec_code=False, network=False)

Bases: object

Declarative definition of one agent-callable tool.

name

Dotted, namespaced, unique tool name (e.g. "compile.dag").

Type:

str

description

One/two-sentence description for the agent (what + when to call).

Type:

str

schema

JSON Schema (draft-07 style {"type": "object", "properties": {...}, "required": [...]}) describing the arguments.

Type:

Dict[str, Any]

handler

Callable taking the validated arg dict, returning a ToolResult.

Type:

Callable[[Dict[str, Any]], cursus.mcp.envelope.ToolResult]

namespace

Leading segment of name (derived; "compile").

destructive

True if the tool mutates external state (e.g. upserts/starts a SageMaker pipeline). Agents/servers may gate these behind confirmation.

Type:

bool

tags

Free-form labels (e.g. "planner", "validator") for grouping.

Type:

tuple

when

One-line “call this when …” cue — the trigger condition, complementing the what in description. Optional; surfaced by help/describe and exporters.

Type:

str

examples

Copy-paste invocation strings showing real calls, e.g. 'catalog.list_steps {"framework": "xgboost"}  # every XGBoost step'. Stored as a tuple so this frozen dataclass field is truly immutable (a list could be mutated in place); surfaced by help/describe and folded into exported tool descriptions so external MCP/OpenAI clients see them too.

Type:

tuple

destructive: bool = False

examples: tuple = ()

exec_code: bool = False

property namespace: str

network: bool = False

tags: tuple = ()

when: str = ''

property wire_name: str

The on-the-wire tool name for MCP hosts (dotted . -> __).

Host tool-calling APIs (Anthropic ^[a-zA-Z0-9_-]{1,128}$, OpenAI ^[a-zA-Z0-9_-]+$) reject the . in the internal dotted names, so every externally-exposed name uses __ instead. It round-trips unambiguously — no tool or namespace name contains a literal __ — via get_tool().

writes: bool = False

name: str

description: str

schema: Dict[str, Any]

handler: Callable[[Dict[str, Any]], ToolResult]

get_registry(force_reload=False)

Return the canonical name -> ToolDef map, building it once and caching.

get_namespaces()

Return the namespace -> one-line description map (built with the registry).

list_tools(namespace=None)

List all registered tools, optionally filtered to one namespace.

get_tool(name)

Look up a tool by its internal dotted name OR its on-the-wire __ name.

call_tool(name, args=None)

Invoke a registered tool by name with an argument dict.

Returns a ToolResult in all cases — unknown tool, invalid args, handled ToolError, or unexpected exception are all converted to error envelopes.

render_description(td)

Compose a tool’s full-text description: description + ‘When’ + ‘Examples’.

External MCP/OpenAI clients only receive a single description string per tool, so the when cue and examples are folded in here (they are otherwise only reachable via the in-process help/describe tools). In-process agents get the fields structured; this keeps external agents from missing the usage guidance.

export_openai_tools(namespace=None)

Export tools in OpenAI / Claude function-calling shape.

The phase tags (planner/validator/programmer) are surfaced under the function’s metadata so an agent can group/route tools by lifecycle phase rather than scanning every description. The when/examples guidance is folded into the description (see render_description()) so external clients see it too.

export_mcp_tools(namespace=None)

Export tools in MCP list_tools shape (name / description / inputSchema).

Includes the phase tags so agents can filter by planner/validator/programmer, and folds the when/examples guidance into the description (see render_description()) so external MCP clients see it too.

Modules

envelope	The result envelope every cursus MCP tool returns.
registry	The canonical tool registry for cursus.mcp.
server	Optional MCP server adapter for the cursus toolset.
tools	Tool namespaces for cursus.mcp.