omicverse.mcp.executor¶
source module omicverse.mcp.executor
Unified execution coordinator for MCP tool calls.
Handles the full lifecycle: resolve → validate → check prerequisites → select adapter → invoke → return structured response.
Classes
-
McpExecutor — Orchestrates MCP tool execution with validation and error handling.
source class McpExecutor(manifest: List[dict], store: SessionStore, adapters: Optional[List[BaseAdapter]] = None)
Orchestrates MCP tool execution with validation and error handling.
Methods
-
execute_tool — Execute a tool by canonical name.
-
resolve_entry — Look up a manifest entry by canonical tool name.
-
validate_call_args — Validate arguments against the parameter schema.
-
check_prerequisites — Check function prerequisites against actual AnnData state.
-
check_data_requirements — Check data structure requirements.
-
select_adapter — Pick the adapter that handles this entry's execution class.
-
suggest_next_tools — Generate suggested next tools based on error context.
source property McpExecutor.store: SessionStore
source property McpExecutor.manifest: List[dict]
source method McpExecutor.execute_tool(tool_name: str, args: dict) → dict
Execute a tool by canonical name.
Returns a response envelope (always a dict with ok key).
Records tool_called / tool_failed events for observability.
source method McpExecutor.resolve_entry(tool_name: str) → Optional[dict]
Look up a manifest entry by canonical tool name.
source method McpExecutor.validate_call_args(entry: dict, args: dict) → Optional[dict]
Validate arguments against the parameter schema.
Returns None if valid, or an error response dict.
source method McpExecutor.check_prerequisites(entry: dict, args: dict) → Optional[dict]
Check function prerequisites against actual AnnData state.
Only checks requires — validates that required data structures
exist on the AnnData object. Does NOT track call history.
source method McpExecutor.check_data_requirements(entry: dict, args: dict) → Optional[dict]
Check data structure requirements.
This is a separate check from prerequisites for clarity, but in
practice they overlap. This method checks requires keys on
the actual AnnData.
source method McpExecutor.select_adapter(entry: dict) → Optional[BaseAdapter]
Pick the adapter that handles this entry's execution class.
source method McpExecutor.build_success_response(tool_name: str, summary: str, outputs: List[dict], state_updates: Optional[dict] = None, warnings_list: Optional[List[str]] = None) → dict
source method McpExecutor.build_error_response(error_code: str, message: str, details: Optional[dict] = None, suggested_next_tools: Optional[List[str]] = None) → dict
source method McpExecutor.suggest_next_tools(entry: dict, error_code: str) → List[str]
Generate suggested next tools based on error context.