Add ManageEngineAgent

patchwork/common/tools/__init__.py

@@ -1,6 +1,7 @@
 from patchwork.common.tools.bash_tool import BashTool
 from patchwork.common.tools.code_edit_tools import CodeEditTool, FileViewTool
 from patchwork.common.tools.grep_tool import FindTextTool, FindTool
+from patchwork.common.tools.api_tool import APIRequestTool
 from patchwork.common.tools.tool import Tool
 
 __all__ = [
@@ -10,4 +11,5 @@
     "FileViewTool",
     "FindTool",
     "FindTextTool",
+    "APIRequestTool",
 ]

patchwork/common/tools/api_tool.py

@@ -0,0 +1,109 @@
+import json
+from typing import Any, Callable, Dict, Optional
+
+import requests
+from typing_extensions import Literal
+
+from patchwork.common.tools.tool import Tool
+
+
+class APIRequestTool(Tool, tool_name="make_api_request", abc_register=False):
+    def __init__(
+        self,
+        headers: Optional[Dict[str, str]] = dict(),
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        preprocess_data: Callable[[str], str] = lambda x: x,
+        **kwargs,
+    ):
+        self._headers = headers
+        self._auth = (username, password) if username and password else None
+        self._preprocess_data = preprocess_data
+
+    @property
+    def json_schema(self) -> dict:
+        return {
+            "name": "make_api_request",
+            "description": """\
+A generic tool to make HTTP API requests with flexible configuration.
+
+Supports various HTTP methods (GET, POST, PUT, DELETE, PATCH) with optional
+authentication, headers, query parameters, and request body.
+
+Authentication can be configured via:
+- Basic Auth (username/password)
+- Bearer Token
+- API Key (in header or query param)
+""",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "url": {
+                        "type": "string",
+                        "description": "Full URL for the API endpoint",
+                    },
+                    "method": {
+                        "type": "string",
+                        "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"],
+                        "description": "HTTP method for the request",
+                    },
+                    "headers": {
+                        "type": "object",
+                        "description": "Optional custom headers",
+                    },
+                    "params": {
+                        "type": "object",
+                        "description": "Optional query parameters",
+                    },
+                    "data": {
+                        "type": "string",
+                        "description": "data for POST/PUT/PATCH requests. If you need to send json data, it should be converted to a string.",
+                    },
+                },
+                "required": ["url", "method"],
+            },
+        }
+
+    def execute(
+        self,
+        url: str,
+        method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"] = "GET",
+        headers: Optional[Dict[str, str]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        data: Optional[str] = None,
+    ) -> str:
+        # Combine with default headers
+        request_headers = headers or {}
+        request_headers.update(self._headers)
+
+        # Prepare request
+        response = requests.request(
+            method=method,
+            url=url,
+            headers=request_headers,
+            params=params,
+            data=(self._preprocess_data(data) if data else None),
+            auth=self._auth,
+        )
+
+        if not response.ok:
+            response_text = response.text
+            status_code = response.status_code
+            headers = response.headers
+
+            header_string = "\n".join(
+                f"{key}: {value}" for key, value in headers.items()
+            )
+
+            return (
+                f"HTTP/{response.raw.version / 10:.1f} {status_code} {response.reason}\n"
+                f"{header_string}\n"
+                f"\n"
+                f"{response_text}"
+            )
+
+        # Try to parse JSON, fallback to text
+        try:
+            return json.dumps(response.json())
+        except ValueError:
+            return response.text

patchwork/steps/BrowserUse/BrowserUse.py

@@ -3,6 +3,7 @@
 import os
 from datetime import datetime
 
+from patchwork.common.utils.utils import mustache_render
 from patchwork.step import Step
 from patchwork.steps import SimplifiedLLMOnce
 from patchwork.steps.BrowserUse.typed import BrowserUseInputs, BrowserUseOutputs
@@ -178,7 +179,7 @@ def run(self) -> dict:
         agent = Agent(
             browser=browser,
             controller=controller,
-            task=self.inputs["task"],
+            task=mustache_render(self.inputs["task"], self.inputs["task_value"]),
             llm=self.llm,
             generate_gif=self.generate_gif,
             validate_output=True,

patchwork/steps/BrowserUse/typed.py

@@ -1,21 +1,19 @@
-from typing_extensions import Annotated, TypedDict
+from typing_extensions import Annotated, Any, Dict, Optional, TypedDict
 
 from patchwork.common.utils.step_typing import StepTypeConfig
 
 
-class BrowserUseInputs(TypedDict, total=False):
+class __BrowserUseInputsRequired(TypedDict):
     task: str
-    example_json: str
-    openai_api_key: Annotated[
-        str,
-        StepTypeConfig(is_config=True, or_op=["google_api_key", "anthropic_api_key"]),
-    ]
-    anthropic_api_key: Annotated[str, StepTypeConfig(is_config=True, or_op=["google_api_key", "openai_api_key"])]
-    google_api_key: Annotated[
-        str,
-        StepTypeConfig(is_config=True, or_op=["openai_api_key", "anthropic_api_key"]),
-    ]
-    generate_gif: Annotated[bool, StepTypeConfig(is_config=True)]
+    task_value: Dict[str, Any]
+
+
+class BrowserUseInputs(__BrowserUseInputsRequired, total=False):
+    example_json: Optional[str]
+    openai_api_key: Annotated[str, StepTypeConfig(or_op=["google_api_key", "anthropic_api_key"])]
+    anthropic_api_key: Annotated[str, StepTypeConfig(or_op=["google_api_key", "openai_api_key"])]
+    google_api_key: Annotated[str, StepTypeConfig(or_op=["openai_api_key", "anthropic_api_key"])]
+    generate_gif: Optional[bool]
 
 
 class BrowserUseOutputs(TypedDict):

patchwork/steps/GitHubAgent/GitHubAgent.py

@@ -6,6 +6,7 @@
     AgenticStrategyV2,
 )
 from patchwork.common.tools.github_tool import GitHubTool
+from patchwork.common.utils.utils import mustache_render
 from patchwork.step import Step
 from patchwork.steps.GitHubAgent.typed import GitHubAgentInputs, GitHubAgentOutputs
 
@@ -14,7 +15,8 @@ class GitHubAgent(Step, input_class=GitHubAgentInputs, output_class=GitHubAgentO
     def __init__(self, inputs):
         super().__init__(inputs)
         base_path = inputs.get("base_path", str(Path.cwd()))
-        task = inputs["task"]
+        data = inputs.get("prompt_value", {})
+        task = mustache_render(inputs["task"], data)
         self.agentic_strategy = AgenticStrategyV2(
             model="claude-3-7-sonnet-latest",
             llm_client=AioLlmClient.create_aio_client(inputs),

patchwork/steps/GitHubAgent/typed.py

@@ -3,11 +3,12 @@
 from patchwork.common.utils.step_typing import StepTypeConfig
 
 
-class GitHubAgentInputs(TypedDict, total=False):
+class __GitHubAgentRequiredInputs(TypedDict):
+    task: str
+
+class GitHubAgentInputs(__GitHubAgentRequiredInputs, total=False):
     base_path: str
     prompt_value: Dict[str, Any]
-    system_prompt: str
-    user_prompt: str
     max_llm_calls: Annotated[int, StepTypeConfig(is_config=True)]
     openai_api_key: Annotated[
         str, StepTypeConfig(is_config=True, or_op=["patched_api_key", "google_api_key", "anthropic_api_key"])

patchwork/steps/ManageEngineAgent/ManageEngineAgent.py

@@ -0,0 +1,75 @@
+from patchwork.common.client.llm.aio import AioLlmClient
+from patchwork.common.multiturn_strategy.agentic_strategy_v2 import (
+    AgentConfig,
+    AgenticStrategyV2,
+)
+from patchwork.common.tools.api_tool import APIRequestTool
+from patchwork.common.utils.utils import mustache_render
+from patchwork.step import Step
+
+from .typed import ManageEngineAgentInputs, ManageEngineAgentOutputs
+
+
+class ManageEngineAgent(Step, input_class=ManageEngineAgentInputs, output_class=ManageEngineAgentOutputs):
+    def __init__(self, inputs: dict):
+        super().__init__(inputs)
+
+        if not inputs.get("me_access_token"):
+            raise ValueError("me_access_token is required")
+        if not inputs.get("user_prompt"):
+            raise ValueError("user_prompt is required")
+
+        # Configure conversation limit
+        self.conversation_limit = int(inputs.get("max_agent_calls", 1))
+
+        # Prepare system prompt with ManageEngine context
+        system_prompt = inputs.get(
+            "system_prompt",
+            "Please summarise the conversation given and provide the result in the structure that is asked of you.",
+        )
+
+        self.headers = {
+            "Authorization": f"Zoho-oauthtoken {inputs.get('me_access_token')}",
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/vnd.manageengine.sdp.v3+json",
+        }
+
+        llm_client = AioLlmClient.create_aio_client(inputs)
+
+        # Configure agentic strategy with ManageEngine-specific context
+        self.agentic_strategy = AgenticStrategyV2(
+            model="claude-3-7-sonnet-latest",
+            llm_client=llm_client,
+            system_prompt_template=system_prompt,
+            template_data={},
+            user_prompt_template=mustache_render(inputs.get("user_prompt"), inputs.get("prompt_value")),
+            agent_configs=[
+                AgentConfig(
+                    name="ManageEngine Assistant",
+                    tool_set=dict(
+                        make_api_request=APIRequestTool(
+                            headers=self.headers,
+                            preprocess_data=lambda x: f"input_data={x}",
+                        )
+                    ),
+                    system_prompt="""\
+You are an senior software developer helping the program manager to interact with ManageEngine ServiceDesk via the ServiceDeskPlus API.
+Your goal is to retrieve, create, or modify service desk tickets and related information.
+Use the `make_api_request` tool to interact with the ManageEngine API.
+Skip the headers for the api requests as they are already provided.
+The base url for the ServiceDeskPlus API is https://sdpondemand.manageengine.com/app/itdesk/api/v3
+
+For modifying or creating data, the data should be a json string.
+When you have the result of the information user requested, return the response of the final result tool as is.
+""",
+                )
+            ],
+            example_json=inputs.get("example_json"),
+        )
+
+    def run(self) -> dict:
+        # Execute the agentic strategy
+        result = self.agentic_strategy.execute(limit=self.conversation_limit)
+
+        # Return results with usage information
+        return {**result, **self.agentic_strategy.usage()}

patchwork/steps/ManageEngineAgent/typed.py

@@ -0,0 +1,27 @@
+from typing_extensions import Annotated, Any, Dict, List, Optional, TypedDict
+
+from patchwork.common.utils.step_typing import StepTypeConfig
+
+
+class __ManageEngineAgentInputsRequired(TypedDict):
+    me_access_token: str
+    user_prompt: str
+    prompt_value: Dict[str, Any]
+
+
+class ManageEngineAgentInputs(__ManageEngineAgentInputsRequired, total=False):
+    max_agent_calls: int
+    openai_api_key: Annotated[str, StepTypeConfig(or_op=["google_api_key", "anthropic_api_key"])]
+    anthropic_api_key: Annotated[str, StepTypeConfig(or_op=["google_api_key", "openai_api_key"])]
+    google_api_key: Annotated[str, StepTypeConfig(or_op=["openai_api_key", "anthropic_api_key"])]
+
+    # Prompt and strategy configuration
+    system_prompt: Optional[str]
+    example_json: Optional[Dict]
+
+
+class ManageEngineAgentOutputs(TypedDict):
+    conversation_history: List[Dict]
+    tool_records: List[Dict]
+    request_tokens: int
+    response_tokens: int

patchwork/steps/__init__.py

@@ -33,6 +33,7 @@
 from patchwork.steps.GitHubAgent.GitHubAgent import GitHubAgent
 from patchwork.steps.JoinList.JoinList import JoinList
 from patchwork.steps.LLM.LLM import LLM
+from patchwork.steps.ManageEngineAgent.ManageEngineAgent import ManageEngineAgent
 from patchwork.steps.ModifyCode.ModifyCode import ModifyCode
 from patchwork.steps.ModifyCodeOnce.ModifyCodeOnce import ModifyCodeOnce
 from patchwork.steps.PR.PR import PR
@@ -108,4 +109,5 @@
     "JoinListPB",
     "GetTypescriptTypeInfo",
     "BrowserUse",
+    "ManageEngineAgent",
 ]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "patchwork-cli"
-version = "0.0.106"
+version = "0.0.107"
 description = ""
 authors = ["patched.codes"]
 license = "AGPL"