feat: add system prompt cli and envar config option to prepend system message to /api/chat. Add unit test and update README entry (#32)

Author: jonigl

README.md

@@ -59,7 +59,7 @@
 - 💻 **Typer CLI**: Clean command-line interface with configurable options
 - 📊 **Structured Logging**: Uses loguru for comprehensive logging
 - 📦 **PyPI Package**: Easily installable via pip or uv from PyPI
-
+- 🗣️ **System Prompt Configuration**: Allows setting a system prompt for the assistant's behavior
 
 ## Requirements
 
@@ -204,6 +204,10 @@ CORS_ORIGINS="http://localhost:3000,http://localhost:8080,https://app.example.co
   - Can be overridden with `--ollama-url` CLI parameter
   - Useful for Docker deployments and configuration management
   - Example: `OLLAMA_URL=http://192.168.1.100:11434 ollama-mcp-bridge`
+- `SYSTEM_PROMPT`: Optional system prompt to prepend to all forwarded `/api/chat` requests
+  - Can be set via the `SYSTEM_PROMPT` environment variable or `--system-prompt` CLI flag
+  - If provided, the bridge will prepend a system message (role: `system`) to the beginning of the `messages` array for `/api/chat` requests unless the request already starts with a system message.
+  - Example: `SYSTEM_PROMPT="You are a concise assistant." ollama-mcp-bridge`
 
 **CORS Logging:**
 - The bridge logs CORS configuration at startup
@@ -235,6 +239,9 @@ ollama-mcp-bridge --ollama-url http://192.168.1.100:11434
 # Limit tool execution rounds (prevents excessive tool calls)
 ollama-mcp-bridge --max-tool-rounds 5
 
+# Set a system prompt to prepend to all /api/chat requests
+ollama-mcp-bridge --system-prompt "You are a concise assistant."
+
 # Combine options
 ollama-mcp-bridge --config custom.json --host 0.0.0.0 --port 8080 --ollama-url http://remote-ollama:11434 --max-tool-rounds 10
 
@@ -253,10 +260,10 @@ ollama-mcp-bridge --version
 - `--host`: Host to bind the server (default: `0.0.0.0`)
 - `--port`: Port to bind the server (default: `8000`)
 - `--ollama-url`: Ollama server URL (default: `http://localhost:11434`)
-- `--max-tool-rounds`: Maximum tool execution rounds (default: unlimited, can also be set via `MAX_TOOL_ROUNDS` environment variable)
+- `--max-tool-rounds`: Maximum tool execution rounds (default: unlimited)
 - `--reload`: Enable auto-reload during development
 - `--version`: Show version information, check for updates and exit
-
+- `--system-prompt`: Optional system prompt to prepend to `/api/chat` requests (default: none)
 ### API Usage
 
 The API is available at `http://localhost:8000`.

src/ollama_mcp_bridge/lifecycle.py

@@ -24,11 +24,13 @@ async def lifespan(fastapi_app: FastAPI):
         config_file = getattr(fastapi_app.state, 'config_file', 'mcp-config.json')
         ollama_url = getattr(fastapi_app.state, 'ollama_url', 'http://localhost:11434')
         max_tool_rounds = getattr(fastapi_app.state, 'max_tool_rounds', None)
-
         logger.info(f"Starting with config file: {config_file}, Ollama URL: {ollama_url}, Max tool rounds: {max_tool_rounds if max_tool_rounds else 'unlimited'}")
 
+        # Get optional system prompt
+        system_prompt = getattr(fastapi_app.state, 'system_prompt', None)
+
         # Initialize manager and load servers
-        mcp_manager = MCPManager(ollama_url=ollama_url)
+        mcp_manager = MCPManager(ollama_url=ollama_url, system_prompt=system_prompt)
         mcp_manager.max_tool_rounds = max_tool_rounds
         await mcp_manager.load_servers(config_file)
 

src/ollama_mcp_bridge/main.py

@@ -16,6 +16,7 @@ def cli_app(
     port: int = typer.Option(8000, "--port", help="Port to bind to"),
     ollama_url: str = typer.Option(os.getenv("OLLAMA_URL", "http://localhost:11434"), "--ollama-url", help="Ollama server URL"),
     max_tool_rounds: Optional[int] = typer.Option(os.getenv("MAX_TOOL_ROUNDS", None), "--max-tool-rounds", help="Maximum tool execution rounds (default: unlimited)"),
+    system_prompt: Optional[str] = typer.Option(os.getenv("SYSTEM_PROMPT", None), "--system-prompt", help="System prompt to prepend to messages (can also be set with SYSTEM_PROMPT env var)"),
     reload: bool = typer.Option(False, "--reload", help="Enable auto-reload"),
     version: bool = typer.Option(False, "--version", help="Show version information, check for updates and exit"),
 ):
@@ -25,12 +26,13 @@ def cli_app(
         # Check for updates and print if available
         asyncio.run(check_for_updates(__version__, print_message=True))
         raise typer.Exit(0)
-    validate_cli_inputs(config, host, port, ollama_url, max_tool_rounds)
+    validate_cli_inputs(config, host, port, ollama_url, max_tool_rounds, system_prompt)
 
     # Store config in app state so lifespan can access it
     app.state.config_file = config
     app.state.ollama_url = ollama_url
     app.state.max_tool_rounds = max_tool_rounds
+    app.state.system_prompt = system_prompt
 
     logger.info(f"Starting MCP proxy server on {host}:{port}")
     logger.info(f"Using Ollama server: {ollama_url}")

src/ollama_mcp_bridge/mcp_manager.py

@@ -12,7 +12,7 @@
 class MCPManager:
     """Manager for MCP servers, handling tool definitions and session management."""
 
-    def __init__(self, ollama_url: str = "http://localhost:11434"):
+    def __init__(self, ollama_url: str = "http://localhost:11434", system_prompt: str = None):
         """Initialize MCP Manager
 
         Args:
@@ -22,6 +22,8 @@ def __init__(self, ollama_url: str = "http://localhost:11434"):
         self.all_tools: List[dict] = []
         self.exit_stack = AsyncExitStack()
         self.ollama_url = ollama_url
+        # Optional system prompt that can be prepended to messages
+        self.system_prompt = system_prompt
         self.http_client = httpx.AsyncClient()
 
     async def load_servers(self, config_path: str):

src/ollama_mcp_bridge/proxy_service.py

@@ -18,6 +18,20 @@ def __init__(self, mcp_manager: MCPManager):
         self.mcp_manager = mcp_manager
         self.http_client = httpx.AsyncClient(timeout=None)
 
+    def _maybe_prepend_system_prompt(self, messages: list) -> list:
+        """If a system prompt is configured on the MCP manager, ensure it is the first message.
+
+        Does not duplicate if the first message already has role 'system'.
+        """
+        system_prompt = getattr(self.mcp_manager, 'system_prompt', None)
+        if not system_prompt:
+            return messages
+
+        # If messages is empty or first message isn't a system role, prepend
+        if not messages or messages[0].get('role') != 'system':
+            return [{'role': 'system', 'content': system_prompt}] + messages
+        return messages
+
     async def health_check(self) -> Dict[str, Any]:
         """Check the health of the Ollama server and MCP setup"""
         ollama_healthy = await check_ollama_health_async(self.mcp_manager.ollama_url)
@@ -83,6 +97,7 @@ async def _proxy_with_tools_non_streaming(self, endpoint: str, payload: Dict[str
         payload = dict(payload)
         payload["tools"] = self.mcp_manager.all_tools if self.mcp_manager.all_tools else None
         messages = payload.get("messages") or []
+        messages = self._maybe_prepend_system_prompt(messages)
 
         # Get max tool rounds from app state (None means unlimited)
         max_rounds = getattr(self.mcp_manager, 'max_tool_rounds', None)
@@ -124,6 +139,7 @@ async def _proxy_with_tools_streaming(self, endpoint: str, payload: Dict[str, An
         payload = dict(payload)
         payload["tools"] = self.mcp_manager.all_tools if self.mcp_manager.all_tools else None
         messages = list(payload.get("messages") or [])
+        messages = self._maybe_prepend_system_prompt(messages)
 
         async def stream_ollama(payload_to_send):
             async with httpx.AsyncClient(timeout=None) as client:

src/ollama_mcp_bridge/utils.py

@@ -80,8 +80,12 @@ async def iter_ndjson_chunks(chunk_iterator):
         except json.JSONDecodeError as e:
             logger.debug(f"Error parsing trailing NDJSON: {e}")
 
-def validate_cli_inputs(config: str, host: str, port: int, ollama_url: str, max_tool_rounds: int = None):
-    """Validate CLI inputs for config file, host, port, ollama_url, and max_tool_rounds."""
+def validate_cli_inputs(config: str, host: str, port: int, ollama_url: str, max_tool_rounds: int = None, system_prompt: str = None):
+    """Validate CLI inputs for config file, host, port, ollama_url, max_tool_rounds and system_prompt.
+
+    Args:
+        system_prompt: optional system prompt string; if provided, must be a non-empty string and not excessively long.
+    """
     # Validate config file exists
     if not os.path.isfile(config):
         raise BadParameter(f"Config file not found: {config}")
@@ -103,6 +107,17 @@ def validate_cli_inputs(config: str, host: str, port: int, ollama_url: str, max_
     if max_tool_rounds is not None and max_tool_rounds < 1:
         raise BadParameter(f"max_tool_rounds must be at least 1, got {max_tool_rounds}")
 
+    # Validate system_prompt (if provided)
+    if system_prompt is not None:
+        if not isinstance(system_prompt, str):
+            raise BadParameter("system_prompt must be a string")
+        # Reject empty or whitespace-only prompts
+        if not system_prompt.strip():
+            raise BadParameter("system_prompt must be a non-empty string")
+        # Limit length to a reasonable maximum to avoid excessively large payloads
+        if len(system_prompt) > 10000:
+            raise BadParameter("system_prompt is too long (max 10000 characters)")
+
 async def check_for_updates(current_version: str, print_message: bool = False) -> str:
     """
     Check if a newer version of ollama-mcp-bridge is available on PyPI.

tests/test_unit.py

@@ -140,19 +140,19 @@ def test_validate_cli_max_tool_rounds():
         from ollama_mcp_bridge.utils import validate_cli_inputs
 
     # Valid case: None
-    validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", None)
+    validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", None, None)
 
     # Invalid max_tool_rounds (zero)
     from typer import BadParameter
     try:
-        validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", 0)
+        validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", 0, None)
         assert False, "Expected BadParameter for max_tool_rounds=0"
     except BadParameter:
         pass
 
     # Invalid max_tool_rounds (negative)
     try:
-        validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", -1)
+        validate_cli_inputs("mcp-config.json", "0.0.0.0", 8000, "http://localhost:11434", -1, None)
         assert False, "Expected BadParameter for max_tool_rounds=-1"
     except BadParameter:
         pass
@@ -163,3 +163,34 @@ def test_script_installed():
         assert result.returncode == 0
     except (FileNotFoundError, subprocess.SubprocessError) as e:
         assert False, f"Subprocess call failed. Is the script installed? {e}"
+
+
+def test_system_prompt_prepended():
+    """Test that the system prompt configured on MCPManager is prepended to messages."""
+    try:
+        from ollama_mcp_bridge.mcp_manager import MCPManager
+        from ollama_mcp_bridge.proxy_service import ProxyService
+    except ImportError:
+        sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+        from ollama_mcp_bridge.mcp_manager import MCPManager
+        from ollama_mcp_bridge.proxy_service import ProxyService
+
+    mgr = MCPManager(system_prompt="You are a helpful assistant.")
+    ps = ProxyService(mgr)
+
+    # Case: user message only -> system prompt should be prepended
+    messages = [{"role": "user", "content": "Hello"}]
+    out = ps._maybe_prepend_system_prompt(messages)
+    assert out[0]["role"] == "system"
+    assert out[0]["content"] == "You are a helpful assistant."
+
+    # Case: existing system prompt should not be duplicated or replaced
+    messages2 = [{"role": "system", "content": "Existing"}, {"role": "user", "content": "Hi"}]
+    out2 = ps._maybe_prepend_system_prompt(messages2)
+    assert out2[0]["role"] == "system"
+    assert out2[0]["content"] == "Existing"
+
+    # Case: empty messages -> system prompt becomes the only message
+    out3 = ps._maybe_prepend_system_prompt([])
+    assert out3[0]["role"] == "system"
+    assert out3[0]["content"] == "You are a helpful assistant."