server
Kitaru MCP server tools.
This module exposes structured MCP tools for querying and managing Kitaru executions. The server is intentionally thin: it owns FastMCP wiring, transport-specific validation, and delegation into shared interface helpers.
attributemcp= _load_fastmcp_class()('kitaru')func_load_fastmcp_class() -> type[Any]Load the FastMCP class from optional dependencies.
Returns
type[typing.Any]functracked_mcp_tool(func) -> Callable[..., _T]Register a function as a Kitaru MCP tool with automatic analytics tracking.
Combines @mcp.tool() registration with per-call analytics. Every
invocation fires AnalyticsEvent.MCP_TOOL_CALLED with the tool name
(derived from func.__name__) and the success/failure outcome.
paramfuncCallable[..., _T]Returns
collections.abc.Callable[..., kitaru.mcp.server._T]funckitaru_executions_list(status=None, flow=None, stack=None, limit=None) -> list[dict[str, Any]]List executions with optional status/flow/stack filters.
paramstatusstr | None= Noneparamflowstr | None= Noneparamstackstr | None= Noneparamlimitint | None= NoneReturns
list[dict[str, typing.Any]]funckitaru_executions_get(exec_id) -> dict[str, Any]Get detailed information for one execution.
paramexec_idstrReturns
dict[str, typing.Any]funckitaru_executions_latest(status=None, flow=None, stack=None) -> dict[str, Any]Get the most recent execution matching the provided filters.
paramstatusstr | None= Noneparamflowstr | None= Noneparamstackstr | None= NoneReturns
dict[str, typing.Any]funcget_execution_logs(exec_id, checkpoint=None, source='step', limit=200) -> strFetch runtime log entries for a Kitaru execution.
paramexec_idstrparamcheckpointstr | None= Noneparamsourcestr= 'step'paramlimitint= 200Returns
strfunckitaru_executions_run(target, args=None, stack=None) -> dict[str, Any]Start a flow from \<module_or_file>:\<flow_name> target.
paramtargetstrparamargsdict[str, Any] | None= Noneparamstackstr | None= NoneReturns
dict[str, typing.Any]funckitaru_executions_cancel(exec_id) -> dict[str, Any]Cancel one execution and return updated details.
paramexec_idstrReturns
dict[str, typing.Any]funckitaru_executions_input(exec_id, wait, value) -> dict[str, Any]Provide input to a waiting execution and return updated details.
paramexec_idstrparamwaitstrparamvalueAnyReturns
dict[str, typing.Any]funckitaru_executions_retry(exec_id) -> dict[str, Any]Retry one failed execution and return updated details.
paramexec_idstrReturns
dict[str, typing.Any]funckitaru_executions_replay(exec_id, from_, overrides=None, flow_inputs=None) -> dict[str, Any]Replay an execution and return structured replay details.
paramexec_idstrparamfrom_strparamoverridesdict[str, Any] | None= Noneparamflow_inputsdict[str, Any] | None= NoneReturns
dict[str, typing.Any]funckitaru_artifacts_list(exec_id, name=None, kind=None, producing_call=None, limit=None) -> list[dict[str, Any]]List artifact metadata for one execution.
paramexec_idstrparamnamestr | None= Noneparamkindstr | None= Noneparamproducing_callstr | None= Noneparamlimitint | None= NoneReturns
list[dict[str, typing.Any]]funckitaru_artifacts_get(artifact_id) -> dict[str, Any]Get one artifact's metadata and loaded value.
paramartifact_idstrReturns
dict[str, typing.Any]funckitaru_start_local_server(port=None, timeout=60) -> dict[str, Any]Start or connect to the local Kitaru server.
paramportint | None= Noneparamtimeoutint= 60Returns
dict[str, typing.Any]funckitaru_stop_local_server() -> dict[str, Any]Stop the registered local Kitaru server, if one exists.
Returns
dict[str, typing.Any]funckitaru_status() -> dict[str, Any]Return structured status details for the current Kitaru connection.
Returns
dict[str, typing.Any]funckitaru_stacks_list() -> list[dict[str, Any]]List available stacks from the active connection context.
Returns
list[dict[str, typing.Any]]funcmanage_stack(action, name, activate=True, recursive=False, force=False, stack_type='local', artifact_store=None, container_registry=None, cluster=None, region=None, subscription_id=None, resource_group=None, workspace=None, execution_role=None, namespace=None, credentials=None, extra=None, async_mode=False, verify=True) -> dict[str, Any]Create or delete a local, Kubernetes-backed, Vertex AI, SageMaker,
or AzureML stack. async_mode is the MCP equivalent of CLI --async.
paramactionLiteral['create', 'delete']paramnamestrparamactivatebool= Trueparamrecursivebool= Falseparamforcebool= Falseparamstack_typestr= 'local'paramartifact_storestr | None= Noneparamcontainer_registrystr | None= Noneparamclusterstr | None= Noneparamregionstr | None= Noneparamsubscription_idstr | None= Noneparamresource_groupstr | None= Noneparamworkspacestr | None= Noneparamexecution_rolestr | None= Noneparamnamespacestr | None= Noneparamcredentialsstr | None= Noneparamextradict[str, Any] | None= Noneparamasync_modebool= Falseparamverifybool= TrueReturns
dict[str, typing.Any]funcmain() -> NoneEntry point for the kitaru-mcp console script.
Returns
None