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')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_memory_list(scope, scope_type, prefix=None) -> list[dict[str, Any]]List memory entries for one explicit scope.
paramscopestrparamscope_typestrparamprefixstr | None= NoneReturns
list[dict[str, typing.Any]]funckitaru_memory_get(key, scope, scope_type, version=None) -> dict[str, Any] | NoneGet one memory value from one explicit scope.
paramkeystrparamscopestrparamscope_typestrparamversionint | None= NoneReturns
dict[str, typing.Any] | Nonefunckitaru_memory_set(key, value, scope, scope_type) -> dict[str, Any]Write one memory value into one explicit scope.
paramkeystrparamvalueAnyparamscopestrparamscope_typestrReturns
dict[str, typing.Any]funckitaru_memory_delete(key, scope, scope_type) -> dict[str, Any] | NoneSoft-delete one memory key from one explicit scope.
paramkeystrparamscopestrparamscope_typestrReturns
dict[str, typing.Any] | Nonefunckitaru_memory_history(key, scope, scope_type) -> list[dict[str, Any]]Show all versions for one memory key in one explicit scope.
paramkeystrparamscopestrparamscope_typestrReturns
list[dict[str, typing.Any]]funckitaru_memory_purge(key, scope, scope_type, keep=None) -> dict[str, Any]Physically delete old versions of one memory key.
paramkeystrparamscopestrparamscope_typestrparamkeepint | None= NoneReturns
dict[str, typing.Any]funckitaru_memory_purge_scope(scope, scope_type, keep=None, include_deleted=False) -> dict[str, Any]Purge old versions across all keys in one scope.
paramscopestrparamscope_typestrparamkeepint | None= Noneparaminclude_deletedbool= FalseReturns
dict[str, typing.Any]funckitaru_memory_compact(scope, scope_type, key=None, keys=None, source_mode='current', target_key=None, instruction=None, model=None, max_tokens=None) -> dict[str, Any]Summarize memory values using an LLM and write the result.
paramscopestrparamscope_typestrparamkeystr | None= Noneparamkeyslist[str] | None= Noneparamsource_modeLiteral['current', 'history']= 'current'paramtarget_keystr | None= Noneparaminstructionstr | None= Noneparammodelstr | None= Noneparammax_tokensint | None= NoneReturns
dict[str, typing.Any]funckitaru_memory_compaction_log(scope, scope_type) -> list[dict[str, Any]]Show the compaction audit log for one scope.
paramscopestrparamscope_typestrReturns
list[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]funckitaru_info(all=False, all_packages=False, packages=None) -> dict[str, Any]Return detailed environment diagnostics for the current Kitaru setup.
Equivalent to kitaru info --output json. Use all=True for the full
diagnostic including all packages and environment type.
paramallbool= Falseparamall_packagesbool= Falseparampackageslist[str] | None= NoneReturns
dict[str, typing.Any]funckitaru_clean_preview(scope='project') -> dict[str, Any]Preview what kitaru clean \<scope> would delete (dry-run only).
Returns the same payload as kitaru clean \<scope> --dry-run --output json.
This tool is strictly read-only and never performs actual cleanup.
To execute cleanup, use the CLI: kitaru clean \<scope> --yes.
paramscopeLiteral['project', 'global', 'all']= 'project'Returns
dict[str, typing.Any]funcmain() -> NoneEntry point for the kitaru-mcp console script.
Returns
None