feat: add MCP tool orchestration and server config support

Introduce server-side MCP config loading from /etc/pr-agent/mcp.json or MCP_CONFIG_PATH, including JSONC parsing and VS Code / Claude schema normalization. Add the MCP runtime, HTTP and stdio clients, structured tool-calling orchestration on the base AI handler, and wire /ask, /review, and /improve through the MCP-aware integration helper. Expose MCP runtime status in /config output, document the configuration flow and AWS Knowledge example, and add focused tests for config loading, runtime behavior, tool orchestration, integration, and discovery.

Author: universam1

docs/docs/usage-guide/additional_configurations.md

@@ -9,6 +9,27 @@ To print all the available configurations as a comment on your PR, you can use t
 /config
 ```
 
+When MCP is enabled, the `/config` comment also includes a small MCP runtime status block showing whether MCP is enabled and which servers are configured and connected.
+
+## MCP runtime configuration
+
+PR-Agent can load MCP servers from a server-side JSON or JSONC file. By default, it reads `/etc/pr-agent/mcp.json`, and you can override that path with `MCP_CONFIG_PATH` or the `[mcp].config_path` setting.
+
+The file may use either the `servers` key, which matches the VS Code MCP schema, or `mcpServers`, which matches the Claude Desktop schema.
+
+For example, an AWS Knowledge MCP server can be configured like this:
+
+```json
+{
+    "servers": {
+        "AWS Knowledge": {
+            "url": "https://knowledge-mcp.global.api.aws",
+            "type": "http"
+        }
+    }
+}
+```
+
 ![possible_config1](https://codium.ai/images/pr_agent/possible_config1.png){width=512}
 
 To view the **actual** configurations used for a specific tool, after all the user settings are applied, you can add for each tool a `--config.output_relevant_configurations=true` suffix.

docs/docs/usage-guide/automations_and_usage.md

@@ -75,6 +75,8 @@ For example, if you want to edit the `review` tool configurations, you can run:
 
 Any configuration value in [configuration file](https://github.com/the-pr-agent/pr-agent/blob/main/pr_agent/settings/configuration.toml) file can be similarly edited. Comment `/config` to see the list of available configurations.
 
+If you want PR-Agent to use MCP tools, mount a server-side MCP config file at `/etc/pr-agent/mcp.json` or point `MCP_CONFIG_PATH` at another JSON/JSONC file. The `/config` comment will show the active MCP runtime status when MCP is enabled.
+
 ## PR-Agent Automatic Feedback
 
 ### Disabling all automatic feedback

pr_agent/algo/ai_handlers/base_ai_handler.py

@@ -1,4 +1,8 @@
+import inspect
+import json
+import logging
 from abc import ABC, abstractmethod
+from typing import Any, Awaitable, Callable, Optional
 
 
 class BaseAiHandler(ABC):
@@ -10,6 +14,8 @@ class BaseAiHandler(ABC):
     def __init__(self):
         pass
 
+    _logger = logging.getLogger(__name__)
+
     @property
     @abstractmethod
     def deployment_id(self):
@@ -26,3 +32,169 @@ async def chat_completion(self, model: str, system: str, user: str, temperature:
             temperature (float): the temperature to use for the chat completion
         """
         pass
+
+    async def chat_completion_with_tools(
+        self,
+        model: str,
+        system: str,
+        user: str,
+        tools: Optional[list[dict[str, Any]]] = None,
+        tool_executor: Optional[Callable[[str, dict[str, Any]], Any | Awaitable[Any]]] = None,
+        temperature: float = 0.2,
+        img_path: str = None,
+        max_tool_turns: int = 4,
+        max_tool_output_chars: int = 12000,
+    ):
+        """
+        Run a structured tool-calling loop on top of plain chat completion.
+
+        The model is instructed to emit JSON tool requests in the form:
+        {"type": "tool_call", "tool": "server.tool", "arguments": {...}}
+        and to finish with:
+        {"type": "final", "content": "..."}
+
+        max_tool_output_chars is applied per tool call, not across all tool calls.
+        """
+        if not tools or tool_executor is None:
+            return await self.chat_completion(model, system, user, temperature=temperature, img_path=img_path)
+
+        tool_catalog_text = json.dumps(tools, indent=2, sort_keys=True)
+        structured_system = (
+            f"{system}\n\n"
+            f"Available MCP tools (JSON schema):\n{tool_catalog_text}\n\n"
+            "Always inspect the available tools first and use them before responding whenever they can help answer the user's request.\n"
+            "When you need a tool, respond with ONLY a JSON object exactly in this shape:\n"
+            '{"type":"tool_call","tool":"server.tool","arguments":{...}}\n'
+            "Do not include a final answer in the same message as a tool call.\n"
+            "When you are finished, respond with ONLY a JSON object exactly in this shape:\n"
+            '{"type":"final","content":"..."}\n'
+            "Do not wrap the JSON in markdown fences."
+        )
+
+        conversation_history = [user]
+        remaining_turns = max_tool_turns
+        current_img_path = img_path
+
+        while True:
+            current_user = "\n\n".join(conversation_history)
+            response_text, finish_reason = await self.chat_completion(
+                model=model,
+                system=structured_system,
+                user=current_user,
+                temperature=temperature,
+                img_path=current_img_path,
+            )
+            current_img_path = None
+
+            parsed_response = self._parse_tool_or_final_response(response_text)
+            if parsed_response is None:
+                return response_text, finish_reason
+
+            response_type = parsed_response.get("type", "final")
+            if response_type == "final":
+                return str(parsed_response.get("content", "")), finish_reason
+
+            if response_type != "tool_call":
+                return response_text, finish_reason
+
+            if remaining_turns <= 0:
+                raise ValueError("MCP tool orchestration exceeded the configured turn budget")
+
+            tool_name = str(parsed_response.get("tool", "")).strip()
+            arguments = parsed_response.get("arguments") or {}
+            if not tool_name:
+                raise ValueError("MCP tool orchestration returned an empty tool name")
+            if not isinstance(arguments, dict):
+                raise ValueError("MCP tool orchestration arguments must be a JSON object")
+
+            tool_result = tool_executor(tool_name, arguments)
+            if inspect.isawaitable(tool_result):
+                tool_result = await tool_result
+
+            tool_result_text = self._normalize_tool_result_text(
+                tool_result,
+                max_tool_output_chars=max_tool_output_chars,
+                tool_name=tool_name,
+            )
+            conversation_history.append(f"Previous assistant tool request:\n{response_text}")
+            conversation_history.append(f"Tool result for {tool_name}:\n{tool_result_text}")
+            remaining_turns -= 1
+
+    @classmethod
+    def _normalize_tool_result_text(
+        cls,
+        tool_result: Any,
+        max_tool_output_chars: int,
+        tool_name: str = "<unknown>",
+    ) -> str:
+        if isinstance(tool_result, str):
+            result_text = tool_result
+        else:
+            result_text = json.dumps(tool_result, indent=2, sort_keys=True, default=str)
+
+        if len(result_text) > max_tool_output_chars:
+            cls._logger.warning(
+                "Tool output for '%s' exceeded per-tool max_tool_output_chars (%s > %s); truncating output",
+                tool_name,
+                len(result_text),
+                max_tool_output_chars,
+            )
+            suffix = "\n[tool output truncated]"
+            truncated_prefix_len = max(0, max_tool_output_chars - len(suffix))
+            return result_text[:truncated_prefix_len] + suffix
+        return result_text
+
+    @staticmethod
+    def _parse_tool_or_final_response(response_text: str) -> Optional[dict[str, Any]]:
+        candidate = response_text.strip()
+        if not candidate:
+            return None
+
+        for json_candidate in BaseAiHandler._iter_json_object_candidates(candidate):
+            try:
+                parsed = json.loads(json_candidate)
+            except json.JSONDecodeError:
+                continue
+
+            if isinstance(parsed, dict):
+                response_type = parsed.get("type")
+                if response_type in {"tool_call", "final"}:
+                    return parsed
+
+        return None
+
+    @staticmethod
+    def _iter_json_object_candidates(text: str) -> list[str]:
+        candidates: list[str] = []
+        depth = 0
+        start_index: Optional[int] = None
+        in_string = False
+        is_escaped = False
+
+        for index, char in enumerate(text):
+            if in_string:
+                if is_escaped:
+                    is_escaped = False
+                elif char == "\\":
+                    is_escaped = True
+                elif char == '"':
+                    in_string = False
+                continue
+
+            if char == '"':
+                in_string = True
+                continue
+
+            if char == "{":
+                if depth == 0:
+                    start_index = index
+                depth += 1
+                continue
+
+            if char == "}" and depth > 0:
+                depth -= 1
+                if depth == 0 and start_index is not None:
+                    candidates.append(text[start_index : index + 1])
+                    start_index = None
+
+        return candidates

pr_agent/config_loader.py

@@ -1,11 +1,15 @@
+import json
+import os
 from os.path import abspath, dirname, join
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional
 
 from dynaconf import Dynaconf
 from starlette_context import context
 
 PR_AGENT_TOML_KEY = 'pr-agent'
+MCP_CONFIG_ENV_VAR = "MCP_CONFIG_PATH"
+DEFAULT_MCP_CONFIG_PATH = "/etc/pr-agent/mcp.json"
 
 current_dir = dirname(abspath(__file__))
 
@@ -60,6 +64,134 @@ def get_settings(use_context=False):
         return global_settings
 
 
+def _get_logger():
+    try:
+        from pr_agent.log import get_logger
+        return get_logger()
+    except Exception:
+        class DummyLogger:
+            def debug(self, *args, **kwargs): pass
+            def info(self, *args, **kwargs): pass
+            def warning(self, *args, **kwargs): pass
+            def error(self, *args, **kwargs): pass
+        return DummyLogger()
+
+
+def _strip_json_comments(content: str) -> str:
+    """Strip line and block comments from JSONC-style config while preserving newlines."""
+    stripped = []
+    in_string = False
+    in_line_comment = False
+    in_block_comment = False
+    is_escaped = False
+    index = 0
+
+    while index < len(content):
+        char = content[index]
+        next_char = content[index + 1] if index + 1 < len(content) else ""
+
+        if in_line_comment:
+            if char == "\n":
+                in_line_comment = False
+                stripped.append(char)
+            index += 1
+            continue
+
+        if in_block_comment:
+            if char == "*" and next_char == "/":
+                in_block_comment = False
+                index += 2
+                continue
+            if char == "\n":
+                stripped.append(char)
+            index += 1
+            continue
+
+        if in_string:
+            stripped.append(char)
+            if is_escaped:
+                is_escaped = False
+            elif char == "\\":
+                is_escaped = True
+            elif char == '"':
+                in_string = False
+            index += 1
+            continue
+
+        if char == '"':
+            in_string = True
+            stripped.append(char)
+            index += 1
+            continue
+
+        if char == "/" and next_char == "/":
+            in_line_comment = True
+            index += 2
+            continue
+
+        if char == "/" and next_char == "*":
+            in_block_comment = True
+            index += 2
+            continue
+
+        stripped.append(char)
+        index += 1
+
+    return "".join(stripped)
+
+
+def _resolve_mcp_config_path() -> Path:
+    env_path = os.getenv(MCP_CONFIG_ENV_VAR)
+    if env_path:
+        return Path(env_path).expanduser()
+    configured_path = get_settings().get("MCP.CONFIG_PATH", DEFAULT_MCP_CONFIG_PATH)
+    return Path(str(configured_path)).expanduser()
+
+
+def _normalize_mcp_servers(config_data: dict[str, Any]) -> dict[str, Any]:
+    servers = config_data.get("servers")
+    if servers is None:
+        servers = config_data.get("mcpServers")
+    if servers is None:
+        raise ValueError("MCP config must define either 'servers' or 'mcpServers'")
+    if not isinstance(servers, dict):
+        raise ValueError("MCP server definitions must be a JSON object")
+    return servers
+
+
+def load_mcp_server_config(config_path: Path) -> dict[str, Any]:
+    if not config_path.is_file():
+        raise FileNotFoundError(f"MCP config file not found: {config_path}")
+    config_text = config_path.read_text(encoding="utf-8")
+    try:
+        config_data = json.loads(_strip_json_comments(config_text))
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Invalid MCP config JSON in {config_path}: {exc}") from exc
+    if not isinstance(config_data, dict):
+        raise ValueError("MCP config root must be a JSON object")
+    servers = _normalize_mcp_servers(config_data)
+    return {"servers": servers}
+
+
+def apply_mcp_server_config():
+    logger = _get_logger()
+    config_path = _resolve_mcp_config_path()
+    if not config_path.exists():
+        logger.debug(f"MCP config file not found, skipping load: {config_path}")
+        return
+    try:
+        config_data = load_mcp_server_config(config_path)
+        settings = get_settings()
+        settings.set("MCP.SERVERS", config_data["servers"], merge=False)
+        settings.set("MCP.SERVER_CONFIG", config_data, merge=False)
+        settings.set("MCP.ACTIVE_CONFIG_PATH", str(config_path), merge=False)
+        logger.info(f"Loaded MCP server configuration from {config_path}")
+    except Exception as exc:
+        logger.error(f"Failed to load MCP server configuration from {config_path}: {exc}")
+        if get_settings().get("MCP.FAIL_ON_INVALID_CONFIG", False):
+            raise
+
+
 # Add local configuration from pyproject.toml of the project being reviewed
 def _find_repository_root() -> Optional[Path]:
     """
@@ -90,6 +222,8 @@ def _find_pyproject() -> Optional[Path]:
 if pyproject_path is not None:
     get_settings().load_file(pyproject_path, env=f'tool.{PR_AGENT_TOML_KEY}')
 
+apply_mcp_server_config()
+
 
 def apply_secrets_manager_config():
     """