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,7 @@
+import inspect
+import json
 from abc import ABC, abstractmethod
+from typing import Any, Awaitable, Callable, Optional
 
 
 class BaseAiHandler(ABC):
@@ -26,3 +29,111 @@ 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": "..."}
+        """
+        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"
+            "When you need a tool, respond with a JSON object exactly in this shape:\n"
+            '{"type":"tool_call","tool":"server.tool","arguments":{...}}\n'
+            "When you are finished, respond with a JSON object exactly in this shape:\n"
+            '{"type":"final","content":"..."}\n'
+            "Do not wrap the JSON in markdown fences."
+        )
+
+        current_user = user
+        remaining_turns = max_tool_turns
+        current_img_path = img_path
+
+        while True:
+            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)
+            current_user = (
+                f"{user}\n\n"
+                f"Previous assistant tool request:\n{response_text}\n\n"
+                f"Tool result for {tool_name}:\n{tool_result_text}"
+            )
+            remaining_turns -= 1
+
+    @staticmethod
+    def _normalize_tool_result_text(tool_result: Any, max_tool_output_chars: int) -> 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:
+            return result_text[: max_tool_output_chars - 20] + "\n[tool output truncated]"
+        return result_text
+
+    @staticmethod
+    def _parse_tool_or_final_response(response_text: str) -> Optional[dict[str, Any]]:
+        candidate = response_text.strip()
+        if not candidate.startswith("{") or not candidate.endswith("}"):
+            return None
+
+        try:
+            parsed = json.loads(candidate)
+        except json.JSONDecodeError:
+            return None
+
+        if not isinstance(parsed, dict):
+            return None
+
+        return parsed

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():
     """

pr_agent/mcp/__init__.py

@@ -0,0 +1,15 @@
+from pr_agent.mcp.runtime import (
+    MCPHttpClient,
+    MCPRuntime,
+    MCPRuntimeError,
+    MCPStdioClient,
+    MCPToolDefinition,
+)
+
+__all__ = [
+    "MCPRuntime",
+    "MCPRuntimeError",
+    "MCPToolDefinition",
+    "MCPStdioClient",
+    "MCPHttpClient",
+]

pr_agent/mcp/integration.py

@@ -0,0 +1,62 @@
+from typing import Any, Optional
+
+from pr_agent.config_loader import get_settings
+from pr_agent.mcp.runtime import MCPRuntime
+
+
+def _get_tool_budget(setting_name: str, default_value: int) -> int:
+    value = get_settings().get(setting_name, default_value)
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default_value
+
+
+async def maybe_chat_completion_with_mcp(
+    ai_handler,
+    model: str,
+    system: str,
+    user: str,
+    temperature: float = 0.2,
+    img_path: str = None,
+    command_name: Optional[str] = None,
+):
+    runtime = MCPRuntime()
+    if not runtime.enabled:
+        return await ai_handler.chat_completion(
+            model=model,
+            system=system,
+            user=user,
+            temperature=temperature,
+            img_path=img_path,
+        )
+
+    max_tools = _get_tool_budget("MCP.MAX_TOOL_CATALOG_TOOLS", 12)
+    max_schema_chars = _get_tool_budget("MCP.MAX_TOOL_CATALOG_SCHEMA_CHARS", 12000)
+
+    tools = runtime.build_tool_schemas(
+        max_tools=max_tools,
+        max_schema_chars=max_schema_chars,
+        include_server_prefix=True,
+    )
+    if not tools:
+        return await ai_handler.chat_completion(
+            model=model,
+            system=system,
+            user=user,
+            temperature=temperature,
+            img_path=img_path,
+        )
+
+    if command_name:
+        system = f"{system}\n\nCommand context: {command_name}"
+
+    return await ai_handler.chat_completion_with_tools(
+        model=model,
+        system=system,
+        user=user,
+        tools=tools,
+        tool_executor=runtime.create_tool_executor(),
+        temperature=temperature,
+        img_path=img_path,
+    )