feat: Add workflow control, task management, event handlers, resources and prompts

conductor_mcp/network/http_proxy.py

@@ -57,3 +57,47 @@ async def http_post(resource_path: str, data: Dict[str, Any] = {}, additional_he
         data=json.dumps(data),
     )
     return response.text
+
+
+async def http_put(resource_path: str, data: Dict[str, Any] = None, additional_headers: Dict[str, str] = {}):
+    """Executes token-authenticated HTTP PUT requests to the provided URL
+
+    :param resource_path: The resource path to apply to the server's API endpoint
+    :param data: A dictionary containing any arguments to pass as part of the PUT request (optional)
+    :param additional_headers: A dictionary containing any additional key/values to add to the headers of the request
+    :return: The results of the PUT request
+    """
+    full_url = os.path.join(os.environ[CONDUCTOR_SERVER_URL], resource_path)
+    logging.debug(f"Requesting url: {full_url}")
+    token = await token_manager.get_token()
+    headers = {
+        "X-Authorization": token,
+        "Content-Type": "application/json; charset=utf-8",
+        **additional_headers,
+    }
+    if data is not None:
+        response = httpx.put(full_url, headers=headers, data=json.dumps(data))
+    else:
+        response = httpx.put(full_url, headers=headers)
+    return response.text
+
+
+async def http_delete(resource_path: str, additional_headers: Dict[str, str] = {}):
+    """Executes token-authenticated HTTP DELETE requests to the provided URL
+
+    :param resource_path: The resource path to apply to the server's API endpoint
+    :param additional_headers: A dictionary containing any additional key/values to add to the headers of the request
+    :return: The results of the DELETE request
+    """
+    full_url = os.path.join(os.environ[CONDUCTOR_SERVER_URL], resource_path)
+    logging.debug(f"Requesting url: {full_url}")
+    token = await token_manager.get_token()
+    response = httpx.delete(
+        full_url,
+        headers={
+            "X-Authorization": token,
+            "Content-Type": "application/json; charset=utf-8",
+            **additional_headers,
+        },
+    )
+    return response.text

conductor_mcp/prompts/__init__.py

@@ -0,0 +1,8 @@
+#  Copyright 2025 Orkes Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.

conductor_mcp/prompts/conductor.py

@@ -0,0 +1,136 @@
+#  Copyright 2025 Orkes Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.
+
+"""
+MCP Prompts for Conductor.
+
+Prompts provide reusable templates for common Conductor operations,
+guiding LLMs through troubleshooting, analysis, and workflow management tasks.
+"""
+
+from typing import Optional
+from fastmcp import FastMCP
+
+
+prompt_mcp = FastMCP("Conductor Prompts")
+
+
+@prompt_mcp.prompt()
+def troubleshoot_workflow(workflow_id: str) -> str:
+    """Troubleshoot a failed or stuck workflow.
+
+    Args:
+        workflow_id: The workflow execution ID to troubleshoot
+    """
+    return f"""I need help troubleshooting workflow {workflow_id}. Please:
+1. Get the current status of the workflow using get_workflow_by_id
+2. Identify any failed tasks and their error messages
+3. Check the task logs for detailed errors using get_task_logs
+4. Analyze the workflow definition to understand the expected flow
+5. Suggest possible solutions based on the error messages
+6. Recommend next steps to resolve the issue
+
+Start by fetching the workflow status."""
+
+
+@prompt_mcp.prompt()
+def analyze_failures(workflow_name: Optional[str] = None, hours: int = 24) -> str:
+    """Analyze recent workflow failures and identify patterns.
+
+    Args:
+        workflow_name: Optional workflow name to filter by
+        hours: Number of hours to look back (default: 24)
+    """
+    filter_text = f' for workflow type "{workflow_name}"' if workflow_name else ""
+    return f"""Analyze all failed workflows in the last {hours} hours{filter_text}.
+
+Please:
+1. Use query_workflow_executions to find failed workflows with query: status="FAILED"
+2. For each failed workflow, examine the failed tasks and error messages
+3. Identify common error patterns and group failures by root cause
+4. Determine the most frequently failing tasks
+5. Analyze potential root causes (configuration issues, external dependencies, timeouts, etc.)
+6. Provide recommendations to prevent future failures
+7. Prioritize fixes based on failure frequency and impact
+
+Start by querying for failed workflows."""
+
+
+@prompt_mcp.prompt()
+def create_workflow(workflow_name: str, description: str) -> str:
+    """Guide to create a new workflow definition.
+
+    Args:
+        workflow_name: Name for the new workflow
+        description: Description of what the workflow does
+    """
+    return f"""I want to create a new workflow called "{workflow_name}".
+Description: {description}
+
+Please help me:
+1. First, use get_all_workflows to show me existing workflow definitions as examples
+2. Based on the description, design the workflow structure with appropriate tasks
+3. Guide me through creating the workflow definition JSON
+4. Ensure all task types are valid (SIMPLE, HTTP, INLINE, SWITCH, DO_WHILE, etc.)
+5. Create any necessary task definitions using create_task_definition
+6. Register the workflow definition using create_workflow_definition
+7. Verify the workflow was created successfully
+
+Remember the key rules:
+- taskReferenceName must be unique within the workflow
+- INLINE tasks can only access variables via inputParameters
+- SWITCH tasks cannot have both switchCaseValue and expression
+- Version numbers must be incremented when updating existing workflows
+
+Start by showing me existing workflows for reference."""
+
+
+@prompt_mcp.prompt()
+def monitor_workflow(workflow_id: str) -> str:
+    """Monitor a running workflow execution.
+
+    Args:
+        workflow_id: The workflow execution ID to monitor
+    """
+    return f"""Please monitor workflow {workflow_id} and provide a detailed status report.
+
+Steps:
+1. Use get_workflow_by_id to fetch the current workflow status
+2. List all completed tasks with their execution times
+3. Identify the currently running task(s) and their progress
+4. Show pending tasks that are yet to execute
+5. Check for any warnings or potential issues (slow tasks, retries, etc.)
+6. If the workflow is stuck, investigate possible causes
+7. Estimate time to completion based on task durations (if possible)
+
+Provide a clear summary of the workflow's health and progress."""
+
+
+@prompt_mcp.prompt()
+def optimize_workflow(workflow_name: str) -> str:
+    """Analyze a workflow for optimization opportunities.
+
+    Args:
+        workflow_name: Name of the workflow to optimize
+    """
+    return f"""Analyze the workflow "{workflow_name}" for optimization opportunities.
+
+Please:
+1. Use get_workflow_by_name to fetch the workflow definition
+2. Review the task structure and identify potential improvements:
+   - Tasks that could run in parallel (using FORK/JOIN)
+   - Unnecessary sequential dependencies
+   - Tasks with excessive timeouts
+   - Opportunities for batching similar operations
+3. Check recent executions for performance patterns using query_workflow_executions
+4. Identify bottleneck tasks that take the longest
+5. Suggest specific changes to improve workflow efficiency
+6. Provide before/after estimates if applicable
+
+Start by fetching the workflow definition."""

conductor_mcp/resources/__init__.py

@@ -0,0 +1,8 @@
+#  Copyright 2025 Orkes Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.

conductor_mcp/resources/conductor.py

@@ -0,0 +1,87 @@
+#  Copyright 2025 Orkes Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.
+
+"""
+MCP Resources for Conductor.
+
+Resources provide read-only access to Conductor data, allowing LLMs to query
+workflow definitions, task definitions, and workflow execution states.
+"""
+
+from fastmcp import FastMCP
+from conductor_mcp.network.http_proxy import http_get
+
+
+resource_mcp = FastMCP("Conductor Resources")
+
+
+@resource_mcp.resource("conductor://workflows/definitions")
+async def get_workflow_definitions() -> str:
+    """List of all registered workflow definitions in Conductor.
+
+    Returns metadata about all workflows registered in Conductor, including
+    workflow names, versions, descriptions, and task configurations.
+    """
+    path = "metadata/workflow?short=false&metadata=true"
+    return await http_get(path)
+
+
+@resource_mcp.resource("conductor://tasks/definitions")
+async def get_task_definitions() -> str:
+    """List of all registered task definitions in Conductor.
+
+    Returns metadata about all tasks registered in Conductor, including
+    task names, descriptions, retry policies, and timeout configurations.
+    """
+    path = "metadata/taskdefs?access=READ&metadata=false"
+    return await http_get(path)
+
+
+@resource_mcp.resource("conductor://workflows/running")
+async def get_running_workflows() -> str:
+    """List of currently running workflow executions.
+
+    Returns workflow executions that are currently in RUNNING status,
+    including workflow IDs, names, start times, and current progress.
+    """
+    path = "workflow/search?query=status%3D%22RUNNING%22"
+    return await http_get(path)
+
+
+@resource_mcp.resource("conductor://workflows/failed")
+async def get_failed_workflows() -> str:
+    """List of recently failed workflow executions.
+
+    Returns workflow executions that have failed, including workflow IDs,
+    names, failure reasons, and failed task information.
+    """
+    path = "workflow/search?query=status%3D%22FAILED%22"
+    return await http_get(path)
+
+
+@resource_mcp.resource("conductor://workflows/paused")
+async def get_paused_workflows() -> str:
+    """List of currently paused workflow executions.
+
+    Returns workflow executions that are currently in PAUSED status,
+    including workflow IDs, names, and pause timestamps.
+    """
+    path = "workflow/search?query=status%3D%22PAUSED%22"
+    return await http_get(path)
+
+
+@resource_mcp.resource("conductor://tasks/queue")
+async def get_task_queue_status() -> str:
+    """Current status of all task queues.
+
+    Returns the current queue status for all task types, including
+    the number of pending tasks in each queue.
+    """
+    path = "tasks/queue/all"
+    return await http_get(path)

conductor_mcp/server.py

@@ -15,13 +15,19 @@
 from conductor_mcp import local_development
 from conductor_mcp.tools.task import task_mcp
 from conductor_mcp.tools.workflow import workflow_mcp
+from conductor_mcp.tools.event import event_mcp
+from conductor_mcp.resources.conductor import resource_mcp
+from conductor_mcp.prompts.conductor import prompt_mcp
 import click
 
 from conductor_mcp.utils.constants import CONDUCTOR_SERVER_URL, CONDUCTOR_AUTH_KEY, CONDUCTOR_AUTH_SECRET
 
 mcp = FastMCP("oss-conductor")
-mcp.mount(workflow_mcp)
-mcp.mount(task_mcp)
+mcp.mount("workflow", workflow_mcp)
+mcp.mount("task", task_mcp)
+mcp.mount("event", event_mcp)
+mcp.mount("resource", resource_mcp)
+mcp.mount("prompt", prompt_mcp)
 
 
 @click.command()

conductor_mcp/test/network/http_proxy_test.py

@@ -46,3 +46,46 @@ async def test_http_post(httpx_mock: HTTPXMock, monkeypatch):
     request = httpx_mock.get_request()
     assert request.headers["header1"] == "header1Val"
     assert request.content == b'{"middleEarth": {"shire": "hobbiton"}}'
+
+
+@pytest.mark.asyncio
+async def test_http_put_with_data(httpx_mock: HTTPXMock, monkeypatch):
+    monkeypatch.setenv(CONDUCTOR_SERVER_URL, TEST_URL)
+    monkeypatch.setattr(token_manager, "get_token", mock_token_retriever)
+    httpx_mock.add_response(url=TEST_URL + f"/somegarbageputurl", text="test_put_response")
+
+    result = await http_proxy.http_put("somegarbageputurl", data={"key": "value"})
+
+    assert result == "test_put_response"
+
+    request = httpx_mock.get_request()
+    assert request.method == "PUT"
+    assert request.content == b'{"key": "value"}'
+
+
+@pytest.mark.asyncio
+async def test_http_put_without_data(httpx_mock: HTTPXMock, monkeypatch):
+    monkeypatch.setenv(CONDUCTOR_SERVER_URL, TEST_URL)
+    monkeypatch.setattr(token_manager, "get_token", mock_token_retriever)
+    httpx_mock.add_response(url=TEST_URL + f"/workflow/123/pause", text="paused")
+
+    result = await http_proxy.http_put("workflow/123/pause")
+
+    assert result == "paused"
+
+    request = httpx_mock.get_request()
+    assert request.method == "PUT"
+
+
+@pytest.mark.asyncio
+async def test_http_delete(httpx_mock: HTTPXMock, monkeypatch):
+    monkeypatch.setenv(CONDUCTOR_SERVER_URL, TEST_URL)
+    monkeypatch.setattr(token_manager, "get_token", mock_token_retriever)
+    httpx_mock.add_response(url=TEST_URL + f"/workflow/123", text="deleted")
+
+    result = await http_proxy.http_delete("workflow/123")
+
+    assert result == "deleted"
+
+    request = httpx_mock.get_request()
+    assert request.method == "DELETE"

conductor_mcp/tools/event.py

@@ -0,0 +1,36 @@
+#  Copyright 2025 Orkes Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+#  the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+#  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+#  specific language governing permissions and limitations under the License.
+
+from typing import Optional
+from fastmcp import FastMCP
+from conductor_mcp.network.http_proxy import http_get
+
+
+event_mcp = FastMCP("Event Service")
+
+
+@event_mcp.tool()
+async def get_event_handlers(event: Optional[str] = None, active_only: bool = True) -> str:
+    """Gets all event handlers or filters by event name and active status.
+
+    Event handlers define how Conductor responds to external events. They can trigger
+    workflow executions or perform other actions when specific events occur.
+
+    Args:
+        event: Optional event name to filter by
+        active_only: If True, return only active event handlers (default: True)
+    """
+    params = []
+    if event:
+        params.append(f"event={event}")
+    params.append(f"activeOnly={str(active_only).lower()}")
+
+    query_string = "&".join(params)
+    path = f"event?{query_string}"
+    return await http_get(path)