Cline Support (#463)

* Cline Support

This should be considered experimental until tester more widely
by the community.

I have it working with Anthropic and Ollama so far.

* Remove list check

* start adding lm_studio

* fix message extraaction

* fix secret redaction

* fix readme

fix lint

fix lint

* fix problems with codegate version and cline detection

fix lint

* fix ollama for cline

---------

Co-authored-by: Yolanda Robla <info@ysoft.biz>
Co-authored-by: Yolanda Robla Mota <yolanda@stacklok.com>

Author: lukehinds

README.md

@@ -82,6 +82,16 @@ With Aider, you can choose from two leading AI model providers:
 - 💻 Local LLMs with [Ollama](https://ollama.com/)
 - 🧠 [OpenAI API](https://openai.com/api/)
 
+- **[Cline](https://github.com/cline/cline)**
+
+With Cline, you can choose between differnet leading AI model providers:
+
+- 🤖 [Anthropic API](https://www.anthropic.com/api)
+- 🧠 [OpenAI API](https://openai.com/api/)
+- 💻 [LM Studio](https://lmstudio.ai/)
+- 💻 Local LLMs with [Ollama](https://ollama.com/)
+
+
 ### Privacy first
 
 Unlike E.T., your code never phones home! 🛸 CodeGate is designed with privacy

src/codegate/config.py

@@ -20,6 +20,7 @@
     "anthropic": "https://api.anthropic.com/v1",
     "vllm": "http://localhost:8000",  # Base URL without /v1 path
     "ollama": "http://localhost:11434",  # Default Ollama server URL
+    "lm_studio": "http://localhost:1234",
 }
 
 

src/codegate/pipeline/cli/cli.py

@@ -1,3 +1,4 @@
+import re
 import shlex
 
 from litellm import ChatCompletionRequest
@@ -76,12 +77,19 @@ async def process(
 
         if last_user_message is not None:
             last_user_message_str, _ = last_user_message
-            splitted_message = last_user_message_str.lower().split(" ")
+            cleaned_message_str = re.sub(r"<.*?>", "", last_user_message_str).strip()
+            splitted_message = cleaned_message_str.lower().split(" ")
             # We expect codegate as the first word in the message
             if splitted_message[0] == "codegate":
                 context.shortcut_response = True
-                args = shlex.split(last_user_message_str)
+                args = shlex.split(cleaned_message_str)
                 cmd_out = await codegate_cli(args[1:])
+
+                if cleaned_message_str != last_user_message_str:
+                    # it came from Cline, need to wrap into tags
+                    cmd_out = (
+                        f"<attempt_completion><result>{cmd_out}</result></attempt_completion>\n"
+                    )
                 return PipelineResult(
                     response=PipelineResponse(
                         step_name=self.name,

src/codegate/pipeline/codegate_context_retriever/codegate.py

@@ -36,7 +36,7 @@ def generate_context_str(self, objects: list[object], context: PipelineContext)
         matched_packages = []
         for obj in objects:
             # The object is already a dictionary with 'properties'
-            package_obj = obj["properties"]
+            package_obj = obj["properties"]  # type: ignore
             matched_packages.append(f"{package_obj['name']} ({package_obj['type']})")
             # Add one alert for each package found
             context.add_alert(
@@ -91,13 +91,16 @@ async def process(
             )  # type: ignore
             logger.info(f"Found {len(bad_snippet_packages)} bad packages in code snippets.")
 
-        # Remove code snippets from the user messages and search for bad packages
+        # Remove code snippets and file listing from the user messages and search for bad packages
         # in the rest of the user query/messsages
         user_messages = re.sub(r"```.*?```", "", user_message, flags=re.DOTALL)
         user_messages = re.sub(r"⋮...*?⋮...\n\n", "", user_messages, flags=re.DOTALL)
+        user_messages = re.sub(
+            r"<environment_details>.*?</environment_details>", "", user_messages, flags=re.DOTALL
+        )
 
         # split messages into double newlines, to avoid passing so many content in the search
-        split_messages = user_messages.split("\n\n")
+        split_messages = re.split(r"</?task>|(\n\n)", user_messages)
         collected_bad_packages = []
         for item_message in split_messages:
             # Vector search to find bad packages
@@ -126,10 +129,26 @@ async def process(
             # Make a copy of the request
             new_request = request.copy()
 
-            # Add the context to the last user message
             # Format: "Context: {context_str} \n Query: {last user message content}"
             message = new_request["messages"][last_user_idx]
-            context_msg = f'Context: {context_str} \n\n Query: {message["content"]}'  # type: ignore
+            message_str = str(message["content"])  # type: ignore
+            # Add the context to the last user message
+            if message_str.strip().startswith("<task>"):
+                # formatting of cline
+                match = re.match(r"(<task>)(.*?)(</task>)(.*)", message_str, re.DOTALL)
+                if match:
+                    task_start, task_content, task_end, rest_of_message = match.groups()
+
+                # Embed the context into the task block
+                updated_task_content = (
+                    f"{task_start}Context: {context_str}\n"
+                    + f"Query: {task_content.strip()}</details>{task_end}"
+                )
+
+                # Combine the updated task block with the rest of the message
+                context_msg = updated_task_content + rest_of_message
+            else:
+                context_msg = f"Context: {context_str} \n\n Query: {message_str}"  # type: ignore
             message["content"] = context_msg
 
             logger.debug("Final context message", context_message=context_msg)

src/codegate/pipeline/extract_snippets/extract_snippets.py

@@ -125,7 +125,8 @@ def extract_snippets(message: str) -> List[CodeSnippet]:
 
         #  just correct the typescript exception
         lang_map = {"typescript": "javascript"}
-        lang = lang_map.get(lang, lang)
+        if lang:
+            lang = lang_map.get(lang, lang)
         snippets.append(CodeSnippet(filepath=filename, code=content, language=lang))
 
     return snippets

src/codegate/pipeline/secrets/secrets.py

@@ -451,17 +451,31 @@ async def process_chunk(
         ):
             return [chunk]
 
+        is_cline_client = any(
+            "Cline" in str(message.trigger_string or "")
+            for message in input_context.alerts_raised or []
+        )
+
         # Check if this is the first chunk (delta role will be present, others will not)
         if len(chunk.choices) > 0 and chunk.choices[0].delta.role:
             redacted_count = input_context.metadata["redacted_secrets_count"]
             secret_text = "secret" if redacted_count == 1 else "secrets"
             # Create notification chunk
-            notification_chunk = self._create_chunk(
-                chunk,
-                f"\n🛡️ [CodeGate prevented {redacted_count} {secret_text}]"
-                f"(http://localhost:9090/?search=codegate-secrets) from being leaked "
-                f"by redacting them.\n\n",
-            )
+            if is_cline_client:
+                notification_chunk = self._create_chunk(
+                    chunk,
+                    f"<thinking>\n🛡️ [CodeGate prevented {redacted_count} {secret_text}]"
+                    f"(http://localhost:9090/?search=codegate-secrets) from being leaked "
+                    f"by redacting them.</thinking>\n\n",
+                )
+                notification_chunk.choices[0].delta.role = "assistant"
+            else:
+                notification_chunk = self._create_chunk(
+                    chunk,
+                    f"\n🛡️ [CodeGate prevented {redacted_count} {secret_text}]"
+                    f"(http://localhost:9090/?search=codegate-secrets) from being leaked "
+                    f"by redacting them.\n\n",
+                )
 
             # Reset the counter
             input_context.metadata["redacted_secrets_count"] = 0

src/codegate/pipeline/secrets/signatures.py

@@ -2,7 +2,7 @@
 import re
 from pathlib import Path
 from threading import Lock
-from typing import ClassVar, Dict, List, NamedTuple, Optional
+from typing import ClassVar, Dict, List, NamedTuple, Optional, Union
 
 import structlog
 import yaml
@@ -215,16 +215,22 @@ def _load_signatures(cls) -> None:
             raise
 
     @classmethod
-    def find_in_string(cls, text: str) -> List[Match]:
-        """Search for secrets in the provided string."""
+    def find_in_string(cls, text: Union[str, List[str]]) -> List[Match]:
+        """Search for secrets in the provided string or list of strings."""
         if not text:
             return []
 
         if not cls._yaml_path:
             raise RuntimeError("SecretFinder not initialized.")
 
         matches = []
-        lines = text.splitlines()
+
+        # Split text into lines for processing
+        try:
+            lines = text.splitlines()
+        except Exception as e:
+            logger.warning(f"Error splitting text into lines: {e}")
+            return []
 
         for line_num, line in enumerate(lines, start=1):
             for group in cls._signature_groups:

src/codegate/pipeline/systemmsg.py

@@ -16,6 +16,7 @@ def get_existing_system_message(request: ChatCompletionRequest) -> Optional[dict
     Returns:
         The existing system message if found, otherwise None.
     """
+
     for message in request.get("messages", []):
         if message["role"] == "system":
             return message
@@ -50,8 +51,18 @@ def add_or_update_system_message(
         context.add_alert("add-system-message", trigger_string=json.dumps(system_message))
         new_request["messages"].insert(0, system_message)
     else:
+        # Handle both string and list content types (needed for Cline (sends list)
+        existing_content = request_system_message["content"]
+        new_content = system_message["content"]
+
+        # Convert list to string if necessary (needed for Cline (sends list)
+        if isinstance(existing_content, list):
+            existing_content = "\n".join(str(item) for item in existing_content)
+        if isinstance(new_content, list):
+            new_content = "\n".join(str(item) for item in new_content)
+
         # Update existing system message
-        updated_content = request_system_message["content"] + "\n\n" + system_message["content"]
+        updated_content = existing_content + "\n\n" + new_content
         context.add_alert("update-system-message", trigger_string=updated_content)
         request_system_message["content"] = updated_content
 

src/codegate/providers/anthropic/completion_handler.py

@@ -16,6 +16,7 @@ async def execute_completion(
         api_key: Optional[str],
         stream: bool = False,
         is_fim_request: bool = False,
+        base_tool: Optional[str] = "",
     ) -> Union[ModelResponse, AsyncIterator[ModelResponse]]:
         """
         Ensures the model name is prefixed with 'anthropic/' to explicitly route to Anthropic's API.

src/codegate/providers/anthropic/provider.py

@@ -32,9 +32,14 @@ def _setup_routes(self):
         Sets up the /messages route for the provider as expected by the Anthropic
         API. Extracts the API key from the "x-api-key" header and passes it to the
         completion handler.
+
+        There are two routes:
+        - /messages: This is the route that is used by the Anthropic API with Continue.dev
+        - /v1/messages: This is the route that is used by the Anthropic API with Cline
         """
 
         @self.router.post(f"/{self.provider_route_name}/messages")
+        @self.router.post(f"/{self.provider_route_name}/v1/messages")
         async def create_message(
             request: Request,
             x_api_key: str = Header(None),