Skip to content

Code reference

Generated from the docstrings in src/ollama_mcp_bridge/. For the narrative version of how these fit together, see Architecture.

CLI

ollama_mcp_bridge.main

Simple CLI entry point for MCP Proxy

cli_app 

cli_app(
    config: str = typer.Option(
        "mcp-config.json",
        "--config",
        help="Path to MCP config JSON file",
    ),
    host: str = typer.Option(
        "0.0.0.0", "--host", help="Host to bind to"
    ),
    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",
    ),
    upstream_header: List[str] = typer.Option(
        [],
        "--upstream-header",
        help="Header to send to the upstream server as 'Name: Value' (repeatable). Can also be set via the UPSTREAM_HEADERS env var (JSON object).",
    ),
    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",
    ),
)

Start the API proxy server with Ollama REST API compatibility and MCP tool integration

Source code in src/ollama_mcp_bridge/main.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def cli_app(
    config: str = typer.Option("mcp-config.json", "--config", help="Path to MCP config JSON file"),
    host: str = typer.Option("0.0.0.0", "--host", help="Host to bind to"),
    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"
    ),
    upstream_header: List[str] = typer.Option(
        [],
        "--upstream-header",
        help="Header to send to the upstream server as 'Name: Value' (repeatable). "
        "Can also be set via the UPSTREAM_HEADERS env var (JSON object).",
    ),
    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"),
):
    """Start the API proxy server with Ollama REST API compatibility and MCP tool integration"""
    if version:
        typer.echo(
            f"{typer.style('ollama-mcp-bridge', fg=typer.colors.BRIGHT_YELLOW, bold=True)} {typer.style('v'+__version__, fg=typer.colors.BRIGHT_CYAN, bold=True)}"
        )
        # 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, system_prompt)

    upstream_headers = parse_upstream_headers(os.getenv("UPSTREAM_HEADERS"), upstream_header)

    # Check if port is available and host is valid before starting
    has_error, error_msg = is_port_in_use(host, port)
    if has_error:
        logger.error(error_msg)
        raise typer.Exit(1)

    # Store config in app state so lifespan can access it
    app.state.config_file = config
    app.state.ollama_url = ollama_url
    app.state.ollama_headers = upstream_headers
    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}")
    logger.info(f"Using config file: {config}")

    # Check for updates (messages will be logged automatically)
    asyncio.run(check_for_updates(__version__))

    # Check Ollama server health before starting
    if not check_ollama_health(ollama_url, headers=upstream_headers):
        logger.info("Please ensure Ollama is running with: ollama serve")
        raise typer.Exit(1)

    # Start the server
    logger.info("API endpoints:")
    logger.info("  • POST /api/chat - Ollama-compatible chat with MCP tools")
    logger.info("  • GET /{path_name} - Transparent proxy to any Ollama endpoint")
    logger.info("  • GET /health - Health check and status")
    logger.info("  • GET /version - Version information and update check")
    logger.info("  • GET /docs - Swagger UI (API documentation)")
    uvicorn.run("ollama_mcp_bridge.api:app", host=host, port=port, reload=reload)

main 

main()

Main entry point for the CLI application

Source code in src/ollama_mcp_bridge/main.py 
88
89
90
def main():
    """Main entry point for the CLI application"""
    typer.run(cli_app)

Application

ollama_mcp_bridge.api

FastAPI application

health async 

health()

Health check endpoint.

Source code in src/ollama_mcp_bridge/api.py 
30
31
32
33
34
35
36
37
38
39
@app.get("/health", summary="Health check", description="Check the health status of the MCP Proxy and Ollama server.")
async def health():
    """Health check endpoint."""
    proxy_service = get_proxy_service()
    if not proxy_service:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Services not initialized")

    health_info = await proxy_service.health_check()
    status_code = status.HTTP_200_OK if health_info["status"] == "healthy" else status.HTTP_503_SERVICE_UNAVAILABLE
    return JSONResponse(status_code=status_code, content=health_info)

chat async 

chat(
    request: Request,
    body: Dict[str, Any] = Body(..., example=CHAT_EXAMPLE),
)

Transparent proxy for Ollama's /api/chat, with MCP tool injection.

Source code in src/ollama_mcp_bridge/api.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@app.post(
    "/api/chat",
    summary="Generate a chat completion",
    description="Transparent proxy to Ollama's /api/chat with MCP tool injection.",
)
async def chat(request: Request, body: Dict[str, Any] = Body(..., example=CHAT_EXAMPLE)):
    """Transparent proxy for Ollama's /api/chat, with MCP tool injection."""
    proxy_service = get_proxy_service()
    if not proxy_service:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Services not initialized")

    try:
        if body.get("stream", False):
            # Starlette owns the receive channel while streaming and already cancels the
            # response generator on disconnect, so the watcher has to stay out of the way
            return await proxy_service.proxy_chat_with_tools(body, stream=True)

        # Buffered: nothing cancels this long await when the client leaves, hence the watcher
        return await run_until_client_disconnects(proxy_service.proxy_chat_with_tools(body, stream=False), request)
    except ClientDisconnected:
        logger.info("/api/chat: client disconnected, aborted the request to Ollama")
        return Response(status_code=CLIENT_CLOSED_REQUEST)
    except httpx.HTTPStatusError as e:
        logger.error(f"/api/chat failed: {e.response.text}")
        raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
    except httpx.RequestError as e:
        logger.error(f"/api/chat connection error: {str(e)}")
        raise HTTPException(status_code=503, detail=f"Could not connect to Ollama server: {str(e)}") from e
    except Exception as e:
        logger.error(f"/api/chat failed: {e}")
        raise HTTPException(status_code=500, detail=f"/api/chat failed: {str(e)}") from e

version async 

version()

Version information endpoint.

Source code in src/ollama_mcp_bridge/api.py 
75
76
77
78
79
80
@app.get("/version", summary="Version information", description="Get version information and check for updates.")
async def version():
    """Version information endpoint."""
    latest_version = await check_for_updates(__version__)

    return {"version": __version__, "latest_version": latest_version}

proxy_to_ollama async 

proxy_to_ollama(request: Request, path_name: str)

Transparent proxy for all other Ollama endpoints.

Source code in src/ollama_mcp_bridge/api.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@app.api_route(
    "/{path_name:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"],
    summary="Transparent proxy",
    description="Transparent proxy to any Ollama endpoint.",
    include_in_schema=False,
)
async def proxy_to_ollama(request: Request, path_name: str):
    """Transparent proxy for all other Ollama endpoints."""
    proxy_service = get_proxy_service()
    if not proxy_service:
        raise HTTPException(status_code=503, detail="Services not initialized")

    try:
        # Must happen before the watcher starts polling the receive channel
        await request.body()

        # This path always buffers, even for stream=true, so the watcher always applies here.
        # If it ever streams, exclude it like /api/chat does: starlette owns the channel then.
        return await run_until_client_disconnects(proxy_service.proxy_generic_request(path_name, request), request)
    # ClientDisconnect is starlette's, raised if the client leaves mid-upload; ClientDisconnected
    # is ours, raised by the watcher. Same outcome: stop working for a client that is gone.
    except (ClientDisconnect, ClientDisconnected):
        logger.info(f"/{path_name}: client disconnected, aborted the request to Ollama")
        return Response(status_code=CLIENT_CLOSED_REQUEST)
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Could not connect to Ollama server: {str(e)}") from e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Proxy request failed: {str(e)}") from e

Lifecycle

ollama_mcp_bridge.lifecycle

Application lifecycle management for FastAPI

lifespan async 

lifespan(fastapi_app: FastAPI)

FastAPI lifespan events

Source code in src/ollama_mcp_bridge/lifecycle.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
    """FastAPI lifespan events"""
    global mcp_manager, proxy_service

    try:
        # Get config from app state with explicit defaults
        config_file = getattr(fastapi_app.state, "config_file", "mcp-config.json")
        ollama_url = getattr(fastapi_app.state, "ollama_url", "http://localhost:11434")
        ollama_headers = getattr(fastapi_app.state, "ollama_headers", None)
        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, system_prompt=system_prompt, ollama_headers=ollama_headers)
        mcp_manager.max_tool_rounds = max_tool_rounds
        await mcp_manager.load_servers(config_file)

        # Initialize services
        proxy_service = ProxyService(mcp_manager)

        # Check for updates (messages will be logged automatically)
        await check_for_updates(__version__)

        logger.success(f"Startup complete. Total tools available: {len(mcp_manager.all_tools)}")
    except (IOError, ValueError, ImportError, httpx.HTTPError) as e:
        logger.error(f"Startup failed: {str(e)}")
        # Reset globals on failed startup
        mcp_manager = None
        proxy_service = None
        raise
    except Exception as e:
        logger.error(f"Unexpected error during startup: {str(e)}")
        mcp_manager = None
        proxy_service = None
        raise

    yield

    # Cleanup on shutdown
    logger.info("Shutting down services...")
    try:
        if proxy_service:
            await proxy_service.cleanup()
    except (IOError, httpx.HTTPError, ConnectionError, TimeoutError) as e:
        logger.error(f"Error during proxy service cleanup: {str(e)}")
    except (ValueError, AttributeError, RuntimeError) as e:
        logger.error(f"Unexpected error during cleanup: {str(e)}")

    try:
        if mcp_manager:
            await mcp_manager.cleanup()
    except (IOError, ConnectionError, TimeoutError) as e:
        logger.error(f"Error during MCP manager cleanup: {str(e)}")

    # Reset globals
    mcp_manager = None
    proxy_service = None
    logger.info("Shutdown complete")

get_mcp_manager 

get_mcp_manager() -> MCPManager

Get the global MCP manager instance.

Source code in src/ollama_mcp_bridge/lifecycle.py 
84
85
86
def get_mcp_manager() -> MCPManager:
    """Get the global MCP manager instance."""
    return mcp_manager

get_proxy_service 

get_proxy_service() -> ProxyService

Get the global proxy service instance.

Source code in src/ollama_mcp_bridge/lifecycle.py 
89
90
91
def get_proxy_service() -> ProxyService:
    """Get the global proxy service instance."""
    return proxy_service

Proxy service

ollama_mcp_bridge.proxy_service

Service for handling proxy requests to Ollama

ProxyService

Service handling all proxy-related operations to Ollama with or without MCP tools

Source code in src/ollama_mcp_bridge/proxy_service.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
class ProxyService:
    """Service handling all proxy-related operations to Ollama with or without MCP tools"""

    def __init__(self, mcp_manager: MCPManager):
        """Initialize the proxy service with an MCP manager."""
        self.mcp_manager = mcp_manager
        self.ollama_headers = dict(getattr(mcp_manager, "ollama_headers", {}) or {})
        is_set, timeout_seconds = get_ollama_proxy_timeout_config()
        # Preserve existing behavior when unset (no timeout for /api/chat). If set, honor it.
        self.http_client = httpx.AsyncClient(timeout=timeout_seconds) if is_set else httpx.AsyncClient(timeout=None)

    def _get_ollama_headers(self, request_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
        """Merge configured Ollama headers with optional forwarded request headers.

        HTTP header names are case-insensitive, but plain dicts are not, so forwarded
        headers are filtered by lowercase name to avoid sending both the forwarded and
        configured value for the same header (e.g. "authorization" and "Authorization").
        """
        configured_names = {name.lower() for name in self.ollama_headers}
        headers = {k: v for k, v in (request_headers or {}).items() if k.lower() not in configured_names}
        headers.update(self.ollama_headers)
        return headers

    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, headers=self.ollama_headers)
        return {
            "status": "healthy" if ollama_healthy else "degraded",
            "ollama_status": "running" if ollama_healthy else "not accessible",
            "tools": len(self.mcp_manager.all_tools),
        }

    async def proxy_chat_with_tools(
        self, payload: Dict[str, Any], stream: bool = False
    ) -> Union[Dict[str, Any], StreamingResponse]:
        """Handle chat requests with potential tool integration

        Args:
            payload: The request payload
            stream: Whether to use streaming response

        Returns:
            Either a dictionary response or a StreamingResponse
        """
        if not await check_ollama_health_async(self.mcp_manager.ollama_url, headers=self.ollama_headers):
            raise httpx.RequestError("Ollama server not accessible", request=None)

        try:
            if stream:
                return StreamingResponse(
                    self._proxy_with_tools_streaming(endpoint="/api/chat", payload=payload),
                    media_type="application/json",
                )
            else:
                return await self._proxy_with_tools_non_streaming(endpoint="/api/chat", payload=payload)
        except httpx.HTTPStatusError as e:
            logger.error(f"Chat proxy failed: {e.response.text}")
            raise
        except httpx.RequestError as e:
            logger.error(f"Chat connection error: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Chat proxy failed: {e}")
            raise

    async def _make_final_llm_call(self, endpoint: str, payload: Dict[str, Any], messages: list) -> Dict[str, Any]:
        """Make a final LLM call without tools to get final answer after tool execution"""
        final_payload = dict(payload)
        final_payload["stream"] = False  # Explicitly disable streaming to get single JSON response
        final_payload["messages"] = messages
        final_payload["tools"] = None  # Don't allow more tool calls
        resp = await self.http_client.post(
            f"{self.mcp_manager.ollama_url}{endpoint}", json=final_payload, headers=self.ollama_headers
        )
        resp.raise_for_status()
        return resp.json()

    async def _stream_final_llm_call(
        self, stream_ollama, payload: Dict[str, Any], messages: list
    ) -> AsyncGenerator[bytes, None]:
        """Stream a final LLM call without tools to get final answer after tool execution"""
        final_payload = dict(payload)
        final_payload["messages"] = messages
        final_payload["tools"] = None  # Don't allow more tool calls

        ndjson_iter = iter_ndjson_chunks(stream_ollama(final_payload))
        async for json_obj in ndjson_iter:
            buffer_chunk = json.dumps(json_obj).encode() + b"\n"
            yield buffer_chunk

    async def _proxy_with_tools_non_streaming(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Handle non-streaming chat requests with tools"""
        payload = dict(payload)
        payload["stream"] = False  # Explicitly disable streaming to get single JSON response
        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)
        current_round = 0

        # Loop to handle potentially multiple rounds of tool calls
        while True:
            # Call Ollama
            current_payload = dict(payload)
            current_payload["messages"] = messages
            resp = await self.http_client.post(
                f"{self.mcp_manager.ollama_url}{endpoint}", json=current_payload, headers=self.ollama_headers
            )
            resp.raise_for_status()
            result = resp.json()

            # Check for tool calls
            tool_calls = self._extract_tool_calls(result)
            if not tool_calls:
                # No more tool calls, return final result
                return result

            # Add assistant's response with tool calls
            response_content = result.get("message", {}).get("content", "")
            messages.append({"role": "assistant", "content": response_content, "tool_calls": tool_calls})

            # Execute tool calls and add results to messages
            messages = await self._handle_tool_calls(messages, tool_calls)

            # Check if we've reached the maximum number of rounds
            current_round += 1
            if max_rounds is not None and current_round >= max_rounds:
                logger.warning(
                    f"Reached maximum tool execution rounds ({max_rounds}), making final LLM call with tool results"
                )
                return await self._make_final_llm_call(endpoint, payload, messages)

            # Continue loop to get next response

    async def _proxy_with_tools_streaming(self, endpoint: str, payload: Dict[str, Any]) -> AsyncGenerator[bytes, None]:
        """Handle streaming chat requests with tools"""

        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:
                async with client.stream(
                    "POST",
                    f"{self.mcp_manager.ollama_url}{endpoint}",
                    json=payload_to_send,
                    headers=self.ollama_headers,
                ) as resp:
                    async for chunk in resp.aiter_bytes():
                        yield chunk

        # Get max tool rounds from app state (None means unlimited)
        max_rounds = getattr(self.mcp_manager, "max_tool_rounds", None)
        current_round = 0

        # Loop to handle potentially multiple rounds of tool calls
        while True:
            current_payload = dict(payload)
            current_payload["messages"] = messages

            response_text = ""
            final_chunk = None
            # Parallel tool calls arrive one per chunk, so collect them, don't replace
            pending_calls: Dict[Any, Any] = {}

            ndjson_iter = iter_ndjson_chunks(stream_ollama(current_payload))
            async for json_obj in ndjson_iter:
                extracted_calls = self._extract_tool_calls(json_obj)
                for tool_call in extracted_calls:
                    pending_calls[tool_call.get("id") or len(pending_calls)] = tool_call

                if json_obj.get("done"):
                    # Hold it back: only the last round's terminal chunk is the client's
                    final_chunk = json_obj
                    break

                message = json_obj.get("message", {})
                # The terminal chunk has no content when streaming, so accumulate it here
                response_text += message.get("content", "") or ""

                if extracted_calls:
                    # The bridge runs these tools itself, so the client must not see them
                    message = {k: v for k, v in message.items() if k != "tool_calls"}
                    json_obj = {**json_obj, "message": message}

                yield json.dumps(json_obj).encode() + b"\n"

            tool_calls = list(pending_calls.values())
            if not tool_calls:
                # No tool calls required, streaming complete
                if final_chunk:
                    yield json.dumps(final_chunk).encode() + b"\n"
                break

            # Tool calls detected; execute them
            messages.append({"role": "assistant", "content": response_text, "tool_calls": tool_calls})
            messages = await self._handle_tool_calls(messages, tool_calls)

            # Check if we've reached the maximum number of rounds
            current_round += 1
            if max_rounds is not None and current_round >= max_rounds:
                logger.warning(
                    f"Reached maximum tool execution rounds ({max_rounds}), making final LLM call with tool results"
                )
                # Stream the final LLM response with tool results (no more tools allowed)
                async for chunk in self._stream_final_llm_call(stream_ollama, payload, messages):
                    yield chunk
                break

    def _extract_tool_calls(self, result: Dict[str, Any]) -> list:
        """Extract tool calls from response"""
        tool_calls = result.get("message", {}).get("tool_calls", [])
        if tool_calls:
            logger.debug(f"Extracted tool_calls from response: {tool_calls}")
        return tool_calls

    async def _handle_tool_calls(self, messages: list, tool_calls: list) -> list:
        """Process tool calls and get results"""
        for tool_call in tool_calls:
            tool_name = tool_call["function"]["name"]
            arguments = tool_call["function"]["arguments"]
            tool_result = await self.mcp_manager.call_tool(tool_name, arguments)
            logger.debug(f"Tool {tool_name} called with args {arguments}, result: {tool_result}")
            messages.append({"role": "tool", "tool_name": tool_name, "content": tool_result})
        return messages

    async def proxy_generic_request(self, path: str, request: Request) -> Response:
        """Proxy any request to Ollama

        Args:
            path: The path to proxy to
            request: The FastAPI request object

        Returns:
            FastAPI Response object
        """
        # Get ollama URL from MCP manager
        ollama_url = self.mcp_manager.ollama_url

        try:
            # Create URL to forward to
            url = f"{ollama_url}/{path}"

            # Copy headers but exclude host
            headers = {k: v for k, v in request.headers.items() if k.lower() != "host"}
            headers = self._get_ollama_headers(headers)

            # Get request body if present
            body = await request.body()

            # Create HTTP client
            is_set, timeout_seconds = get_ollama_proxy_timeout_config()
            client_kwargs = {"timeout": timeout_seconds} if is_set else {}
            async with httpx.AsyncClient(**client_kwargs) as client:
                # Forward the request with the same method
                response = await client.request(
                    request.method, url, headers=headers, params=request.query_params, content=body if body else None
                )

                # Return the response as-is
                return Response(
                    content=response.content, status_code=response.status_code, headers=dict(response.headers)
                )
        except httpx.HTTPStatusError as e:
            logger.error(f"Proxy failed for {path}: {e.response.text}")
            raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
        except httpx.RequestError as e:
            logger.error(f"Proxy connection error for {path}: {str(e)}")
            raise HTTPException(status_code=503, detail=f"Could not connect to target server: {str(e)}") from e
        except Exception as e:
            logger.error(f"Proxy failed for {path}: {e}")
            raise HTTPException(status_code=500, detail=f"Proxy failed: {str(e)}") from e

    async def cleanup(self):
        """Close HTTP client resources"""
        await self.http_client.aclose()

health_check async 

health_check() -> Dict[str, Any]

Check the health of the Ollama server and MCP setup

Source code in src/ollama_mcp_bridge/proxy_service.py 
51
52
53
54
55
56
57
58
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, headers=self.ollama_headers)
    return {
        "status": "healthy" if ollama_healthy else "degraded",
        "ollama_status": "running" if ollama_healthy else "not accessible",
        "tools": len(self.mcp_manager.all_tools),
    }

proxy_chat_with_tools async 

proxy_chat_with_tools(
    payload: Dict[str, Any], stream: bool = False
) -> Union[Dict[str, Any], StreamingResponse]

Handle chat requests with potential tool integration

Parameters:

Name Type Description Default
payload Dict[str, Any]

The request payload

 required
stream bool

Whether to use streaming response

 False

Returns:

Type Description
Union[Dict[str, Any], StreamingResponse]

Either a dictionary response or a StreamingResponse
Source code in src/ollama_mcp_bridge/proxy_service.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def proxy_chat_with_tools(
    self, payload: Dict[str, Any], stream: bool = False
) -> Union[Dict[str, Any], StreamingResponse]:
    """Handle chat requests with potential tool integration

    Args:
        payload: The request payload
        stream: Whether to use streaming response

    Returns:
        Either a dictionary response or a StreamingResponse
    """
    if not await check_ollama_health_async(self.mcp_manager.ollama_url, headers=self.ollama_headers):
        raise httpx.RequestError("Ollama server not accessible", request=None)

    try:
        if stream:
            return StreamingResponse(
                self._proxy_with_tools_streaming(endpoint="/api/chat", payload=payload),
                media_type="application/json",
            )
        else:
            return await self._proxy_with_tools_non_streaming(endpoint="/api/chat", payload=payload)
    except httpx.HTTPStatusError as e:
        logger.error(f"Chat proxy failed: {e.response.text}")
        raise
    except httpx.RequestError as e:
        logger.error(f"Chat connection error: {str(e)}")
        raise
    except Exception as e:
        logger.error(f"Chat proxy failed: {e}")
        raise

proxy_generic_request async 

proxy_generic_request(
    path: str, request: Request
) -> Response

Proxy any request to Ollama

Parameters:

Name Type Description Default
path str

The path to proxy to

 required
request Request

The FastAPI request object

 required

Returns:

Type Description
Response

FastAPI Response object
Source code in src/ollama_mcp_bridge/proxy_service.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
async def proxy_generic_request(self, path: str, request: Request) -> Response:
    """Proxy any request to Ollama

    Args:
        path: The path to proxy to
        request: The FastAPI request object

    Returns:
        FastAPI Response object
    """
    # Get ollama URL from MCP manager
    ollama_url = self.mcp_manager.ollama_url

    try:
        # Create URL to forward to
        url = f"{ollama_url}/{path}"

        # Copy headers but exclude host
        headers = {k: v for k, v in request.headers.items() if k.lower() != "host"}
        headers = self._get_ollama_headers(headers)

        # Get request body if present
        body = await request.body()

        # Create HTTP client
        is_set, timeout_seconds = get_ollama_proxy_timeout_config()
        client_kwargs = {"timeout": timeout_seconds} if is_set else {}
        async with httpx.AsyncClient(**client_kwargs) as client:
            # Forward the request with the same method
            response = await client.request(
                request.method, url, headers=headers, params=request.query_params, content=body if body else None
            )

            # Return the response as-is
            return Response(
                content=response.content, status_code=response.status_code, headers=dict(response.headers)
            )
    except httpx.HTTPStatusError as e:
        logger.error(f"Proxy failed for {path}: {e.response.text}")
        raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
    except httpx.RequestError as e:
        logger.error(f"Proxy connection error for {path}: {str(e)}")
        raise HTTPException(status_code=503, detail=f"Could not connect to target server: {str(e)}") from e
    except Exception as e:
        logger.error(f"Proxy failed for {path}: {e}")
        raise HTTPException(status_code=500, detail=f"Proxy failed: {str(e)}") from e

cleanup async 

cleanup()

Close HTTP client resources

Source code in src/ollama_mcp_bridge/proxy_service.py 
305
306
307
async def cleanup(self):
    """Close HTTP client resources"""
    await self.http_client.aclose()

MCP manager

ollama_mcp_bridge.mcp_manager

MCP Server Management

MCPManager

Manager for MCP servers, handling tool definitions and session management.

Source code in src/ollama_mcp_bridge/mcp_manager.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class MCPManager:
    """Manager for MCP servers, handling tool definitions and session management."""

    def __init__(
        self,
        ollama_url: str = "http://localhost:11434",
        system_prompt: str = None,
        ollama_headers: Optional[Dict[str, str]] = None,
    ):
        """Initialize MCP Manager

        Args:
            ollama_url: URL of the Ollama server
        """
        self.sessions: Dict[str, ClientSession] = {}
        self.all_tools: List[dict] = []
        self.exit_stack = AsyncExitStack()
        self.ollama_url = ollama_url
        self.ollama_headers = ollama_headers or {}
        # Optional system prompt that can be prepended to messages
        self.system_prompt = system_prompt
        is_set, timeout_seconds = get_ollama_proxy_timeout_config()
        self.http_client = httpx.AsyncClient(timeout=timeout_seconds) if is_set else httpx.AsyncClient()

    async def load_servers(self, config_path: str):
        """Load and connect to all MCP servers from config"""
        config_dir = os.path.dirname(os.path.abspath(config_path))
        try:
            with open(config_path, encoding="utf-8") as f:
                config = json.load(f)
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse config file '{config_path}': {e}")
            raise ValueError(f"Invalid JSON in config file '{config_path}': {e}") from e
        except FileNotFoundError:
            logger.error(f"Config file not found: {config_path}")
            raise

        if "mcpServers" not in config:
            logger.error(f"Config file '{config_path}' missing 'mcpServers' key")
            raise ValueError(f"Config file '{config_path}' missing 'mcpServers' key")

        for name, server_config in config["mcpServers"].items():
            resolved_config = dict(server_config)
            resolved_config["cwd"] = config_dir
            await self._connect_server(name, resolved_config)

    async def _connect_server(self, name: str, config: dict):
        """Connect to a single MCP server"""
        server_stack = AsyncExitStack()

        async def _safe_close_stack() -> None:
            try:
                await server_stack.aclose()
            except BaseException as close_error:
                # Some transports can raise ExceptionGroup/RuntimeError while unwinding
                # after a failed connect; avoid crashing startup because of cleanup.
                logger.debug(f"Error cleaning up failed connection for '{name}': {close_error}")

        try:
            # Validate toolFilter configuration if present
            tool_filter = config.get("toolFilter", {})
            if tool_filter:
                mode = tool_filter.get("mode", "include")
                if mode not in ["include", "exclude"]:
                    logger.error(
                        f"Invalid toolFilter mode '{mode}' for server '{name}'. Must be 'include' or 'exclude'."
                    )
                    sys.exit(1)

            # Expand env vars
            cwd = config.get("cwd", os.getcwd())
            config = expand_dict_env_vars(config, cwd)

            if "command" in config:
                params = StdioServerParameters(
                    command=config["command"], args=config.get("args", []), env=config.get("env"), cwd=config.get("cwd")
                )
                transport = await server_stack.enter_async_context(stdio_client(params))
                read, write = transport
            elif "url" in config:
                url = config["url"]
                headers = config.get("headers", {})

                # Determine connection type by URL suffix or default to StreamableHTTP
                if url.rstrip("/").endswith("/sse"):
                    transport = await server_stack.enter_async_context(sse_client(url=url, headers=headers))
                    read, write = transport
                else:
                    # Default to StreamableHTTP if not explicitly /sse
                    transport = await server_stack.enter_async_context(streamablehttp_client(url=url, headers=headers))
                    # streamablehttp_client yields (read, write, get_session_id)
                    read, write, _ = transport
            else:
                raise ValueError(f"Invalid MCP server config for '{name}': must have 'command' or 'url'")

            session = await server_stack.enter_async_context(ClientSession(read, write))
            await session.initialize()
            self.sessions[name] = session
            meta = await session.list_tools()

            # Apply tool filtering if configured
            tool_filter = config.get("toolFilter", {})
            filter_mode = tool_filter.get("mode", "include")
            filter_tools = tool_filter.get("tools", [])

            all_tool_names = [tool.name for tool in meta.tools]
            filtered_tools = []
            found_tools = []
            missing_tools = []
            excluded_tools = []

            if filter_tools:
                if filter_mode == "include":
                    # Include mode: only add tools in the filter list
                    for tool in meta.tools:
                        if tool.name in filter_tools:
                            filtered_tools.append(tool)
                            found_tools.append(tool.name)
                    # Track which filter tools were not found
                    missing_tools = [t for t in filter_tools if t not in all_tool_names]
                elif filter_mode == "exclude":
                    # Exclude mode: add all tools except those in the filter list
                    for tool in meta.tools:
                        if tool.name not in filter_tools:
                            filtered_tools.append(tool)
                        else:
                            excluded_tools.append(tool.name)
            else:
                # No filter or empty filter list: add all tools
                filtered_tools = list(meta.tools)

            # Add filtered tools to the manager
            for tool in filtered_tools:
                tool_def = {
                    "type": "function",
                    "function": {
                        "name": f"{name}.{tool.name}",
                        "description": tool.description,
                        "parameters": tool.inputSchema,
                    },
                    "server": name,
                    "original_name": tool.name,
                }
                self.all_tools.append(tool_def)

            # Transfer ownership of the server stack to the main exit stack
            self.exit_stack.push_async_callback(server_stack.pop_all().aclose)

            # Log connection results with filtering information
            if filter_tools:
                if filter_mode == "include":
                    logger.info(
                        f"Connected to '{name}' with {len(filtered_tools)}/{len(meta.tools)} tools "
                        f"({len(meta.tools) - len(filtered_tools)} filtered)"
                    )
                    if found_tools:
                        logger.info(f"Server '{name}': enabled tools [{', '.join(found_tools)}]")
                    if missing_tools:
                        logger.warning(f"Server '{name}': tools not found in filter [{', '.join(missing_tools)}]")
                elif filter_mode == "exclude":
                    logger.info(
                        f"Connected to '{name}' with {len(filtered_tools)}/{len(meta.tools)} tools "
                        f"({len(excluded_tools)} excluded)"
                    )
                    if excluded_tools:
                        logger.info(f"Server '{name}': excluded tools [{', '.join(excluded_tools)}]")
            else:
                all_tool_names_list = [tool.name for tool in filtered_tools]
                logger.info(f"Connected to '{name}' with {len(meta.tools)} tools")
                if all_tool_names_list:
                    logger.info(f"Server '{name}': available tools [{', '.join(all_tool_names_list)}]")

        except (SystemExit, KeyboardInterrupt):
            await _safe_close_stack()
            raise

        except BaseException as e:
            # This may include CancelledError depending on runtime.
            logger.error(f"Failed to connect to MCP server '{name}': {repr(e)}")
            await _safe_close_stack()

    async def call_tool(self, tool_name: str, arguments: dict):
        """Call a specific tool by name with provided arguments."""
        tool_info = next((t for t in self.all_tools if t["function"]["name"] == tool_name), None)
        if not tool_info:
            raise ValueError(f"Tool {tool_name} not found")
        server_name = tool_info["server"]
        original_name = tool_info["original_name"]
        session = self.sessions[server_name]

        try:
            result = await session.call_tool(original_name, arguments)

            # Defensive extraction of tool result content
            if not result or not hasattr(result, "content"):
                logger.warning(f"Tool {tool_name} returned unexpected result structure: {result}")
                return f"Tool returned an unexpected response format: {str(result)}"

            if not result.content or len(result.content) == 0:
                logger.warning(f"Tool {tool_name} returned empty content")
                return "Tool returned no content"

            # Try to extract text from the first content item
            first_content = result.content[0]

            # Check for 'text' attribute (standard)
            if hasattr(first_content, "text"):
                return first_content.text

            # Fallback: check for other common attributes
            if hasattr(first_content, "data"):
                content = first_content.data
                return json.dumps(content) if isinstance(content, (dict, list)) else str(content)

            if hasattr(first_content, "value"):
                content = first_content.value
                return json.dumps(content) if isinstance(content, (dict, list)) else str(content)

            # Last resort: stringify the content item
            logger.warning(f"Tool {tool_name} content has unexpected structure: {first_content}")
            return str(first_content)

        except Exception as e:
            # Catch validation errors from MCP protocol layer (e.g., Pydantic errors from malformed JSON)
            error_type = type(e).__name__
            error_msg = str(e)
            logger.error(f"Tool {tool_name} execution failed: {error_type}: {error_msg}")

            # Return a formatted error message that the LLM can understand
            return f"Error executing tool: {error_type}: {error_msg}"

    async def cleanup(self):
        """Cleanup all sessions and close HTTP client."""
        await self.http_client.aclose()
        await self.exit_stack.aclose()

load_servers async 

load_servers(config_path: str)

Load and connect to all MCP servers from config

Source code in src/ollama_mcp_bridge/mcp_manager.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def load_servers(self, config_path: str):
    """Load and connect to all MCP servers from config"""
    config_dir = os.path.dirname(os.path.abspath(config_path))
    try:
        with open(config_path, encoding="utf-8") as f:
            config = json.load(f)
    except json.JSONDecodeError as e:
        logger.error(f"Failed to parse config file '{config_path}': {e}")
        raise ValueError(f"Invalid JSON in config file '{config_path}': {e}") from e
    except FileNotFoundError:
        logger.error(f"Config file not found: {config_path}")
        raise

    if "mcpServers" not in config:
        logger.error(f"Config file '{config_path}' missing 'mcpServers' key")
        raise ValueError(f"Config file '{config_path}' missing 'mcpServers' key")

    for name, server_config in config["mcpServers"].items():
        resolved_config = dict(server_config)
        resolved_config["cwd"] = config_dir
        await self._connect_server(name, resolved_config)

call_tool async 

call_tool(tool_name: str, arguments: dict)

Call a specific tool by name with provided arguments.

Source code in src/ollama_mcp_bridge/mcp_manager.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
async def call_tool(self, tool_name: str, arguments: dict):
    """Call a specific tool by name with provided arguments."""
    tool_info = next((t for t in self.all_tools if t["function"]["name"] == tool_name), None)
    if not tool_info:
        raise ValueError(f"Tool {tool_name} not found")
    server_name = tool_info["server"]
    original_name = tool_info["original_name"]
    session = self.sessions[server_name]

    try:
        result = await session.call_tool(original_name, arguments)

        # Defensive extraction of tool result content
        if not result or not hasattr(result, "content"):
            logger.warning(f"Tool {tool_name} returned unexpected result structure: {result}")
            return f"Tool returned an unexpected response format: {str(result)}"

        if not result.content or len(result.content) == 0:
            logger.warning(f"Tool {tool_name} returned empty content")
            return "Tool returned no content"

        # Try to extract text from the first content item
        first_content = result.content[0]

        # Check for 'text' attribute (standard)
        if hasattr(first_content, "text"):
            return first_content.text

        # Fallback: check for other common attributes
        if hasattr(first_content, "data"):
            content = first_content.data
            return json.dumps(content) if isinstance(content, (dict, list)) else str(content)

        if hasattr(first_content, "value"):
            content = first_content.value
            return json.dumps(content) if isinstance(content, (dict, list)) else str(content)

        # Last resort: stringify the content item
        logger.warning(f"Tool {tool_name} content has unexpected structure: {first_content}")
        return str(first_content)

    except Exception as e:
        # Catch validation errors from MCP protocol layer (e.g., Pydantic errors from malformed JSON)
        error_type = type(e).__name__
        error_msg = str(e)
        logger.error(f"Tool {tool_name} execution failed: {error_type}: {error_msg}")

        # Return a formatted error message that the LLM can understand
        return f"Error executing tool: {error_type}: {error_msg}"

cleanup async 

cleanup()

Cleanup all sessions and close HTTP client.

Source code in src/ollama_mcp_bridge/mcp_manager.py 
248
249
250
251
async def cleanup(self):
    """Cleanup all sessions and close HTTP client."""
    await self.http_client.aclose()
    await self.exit_stack.aclose()

Utilities

ollama_mcp_bridge.utils

Utility functions for ollama-mcp-bridge

ClientDisconnected

Bases: Exception

Raised when the client went away before the upstream work finished.

Source code in src/ollama_mcp_bridge/utils.py 
173
174
class ClientDisconnected(Exception):
    """Raised when the client went away before the upstream work finished."""

get_ollama_proxy_timeout_config 

get_ollama_proxy_timeout_config() -> (
    Tuple[bool, Optional[float]]
)

Return (is_set, timeout_seconds) based on OLLAMA_PROXY_TIMEOUT.

  • Unset/empty: (False, None) meaning "do not override"
  • 0: (True, None) meaning "explicitly disable timeout" (warns once)

  • 0: (True, seconds)

Invalid/negative values are ignored with a warning.

Source code in src/ollama_mcp_bridge/utils.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def get_ollama_proxy_timeout_config() -> Tuple[bool, Optional[float]]:
    """Return (is_set, timeout_seconds) based on OLLAMA_PROXY_TIMEOUT.

    - Unset/empty: (False, None) meaning "do not override"
    - 0: (True, None) meaning "explicitly disable timeout" (warns once)
    - >0: (True, seconds)

    Invalid/negative values are ignored with a warning.
    """
    raw = os.getenv(_OLLAMA_PROXY_TIMEOUT_ENV)
    if raw is None:
        return False, None

    raw = raw.strip()
    if not raw:
        return False, None

    try:
        timeout_ms = int(raw)
    except ValueError:
        logger.warning(f"Ignoring {_OLLAMA_PROXY_TIMEOUT_ENV}={raw!r}: expected an integer number of milliseconds.")
        return False, None

    if timeout_ms < 0:
        logger.warning(f"Ignoring {_OLLAMA_PROXY_TIMEOUT_ENV}={timeout_ms}: must be >= 0 (milliseconds).")
        return False, None

    if timeout_ms == 0:
        _warn_ollama_proxy_timeout_disabled_once()
        return True, None

    return True, timeout_ms / 1000.0

is_port_in_use 

is_port_in_use(
    host: str, port: int
) -> Tuple[bool, Optional[str]]

Check if a port is already in use on a given host.

Returns:

Type Description
bool

Tuple[bool, Optional[str]]: (has_error, error_message)
Optional[str]
  • (False, None): Port is available
Tuple[bool, Optional[str]]
  • (True, error_msg): Port check failed with specific error message
Source code in src/ollama_mcp_bridge/utils.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def is_port_in_use(host: str, port: int) -> Tuple[bool, Optional[str]]:
    """Check if a port is already in use on a given host.

    Returns:
        Tuple[bool, Optional[str]]: (has_error, error_message)
        - (False, None): Port is available
        - (True, error_msg): Port check failed with specific error message
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind((host, port))
            return False, None
        except socket.error as e:
            winerror = getattr(e, "winerror", None)
            if e.errno == errno.EADDRINUSE or winerror == 10048:
                return True, f"Port {port} is already in use on {host}. Please use a different port with --port."
            elif e.errno == errno.EACCES or winerror == 10013:
                return (
                    True,
                    f"Port {port} is binding to a privileged port on {host}. Please use a different port with --port.",
                )
            elif e.errno == errno.EADDRNOTAVAIL:
                return True, f"Cannot bind to host '{host}': address not available. Please check the --host value."
            else:
                return True, f"Cannot bind to {host}:{port}: {e.strerror}"

configure_cors 

configure_cors(app)

Configure CORS middleware for the FastAPI app.

Source code in src/ollama_mcp_bridge/utils.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def configure_cors(app):
    """Configure CORS middleware for the FastAPI app."""

    cors_origins = os.getenv("CORS_ORIGINS", "*").split(",")
    cors_origins = [origin.strip() for origin in cors_origins]

    # Don't log CORS config if the user is checking the version
    is_version_check = any("--version" in arg for arg in sys.argv)

    if not is_version_check:
        if cors_origins == ["*"]:
            logger.warning("CORS is configured to allow ALL origins (*). This is not recommended for production.")
        else:
            logger.info(f"CORS configured to allow origins: {cors_origins}")

    app.add_middleware(
        CORSMiddleware,
        allow_origins=cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

check_ollama_health 

check_ollama_health(
    ollama_url: str,
    timeout: int = 3,
    headers: Optional[Dict[str, str]] = None,
) -> bool

Check if Ollama server is running and accessible (sync version for CLI).

Source code in src/ollama_mcp_bridge/utils.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def check_ollama_health(ollama_url: str, timeout: int = 3, headers: Optional[Dict[str, str]] = None) -> bool:
    """Check if Ollama server is running and accessible (sync version for CLI)."""
    try:
        is_set, timeout_override = get_ollama_proxy_timeout_config()
        effective_timeout = timeout_override if is_set else timeout
        resp = httpx.get(f"{ollama_url}/api/tags", timeout=effective_timeout, headers=headers)
        if resp.status_code == 200:
            logger.success("✓ Ollama server is accessible")
            return True
        logger.error(f"Ollama server not accessible at {ollama_url}")
        return False
    except (httpx.ConnectError, httpx.ReadTimeout, httpx.HTTPError) as e:
        logger.error(f"Failed to connect to Ollama: {e}")
        return False

check_ollama_health_async async 

check_ollama_health_async(
    ollama_url: str,
    timeout: int = 3,
    headers: Optional[Dict[str, str]] = None,
) -> bool

Check if Ollama server is running and accessible (async version for FastAPI).

Source code in src/ollama_mcp_bridge/utils.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def check_ollama_health_async(
    ollama_url: str, timeout: int = 3, headers: Optional[Dict[str, str]] = None
) -> bool:
    """Check if Ollama server is running and accessible (async version for FastAPI)."""
    try:
        is_set, timeout_override = get_ollama_proxy_timeout_config()
        effective_timeout = timeout_override if is_set else timeout
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{ollama_url}/api/tags", timeout=effective_timeout, headers=headers)
            if resp.status_code == 200:
                return True
            logger.error(f"Ollama server not accessible at {ollama_url}")
            return False
    except (httpx.ConnectError, httpx.ReadTimeout, httpx.HTTPError) as e:
        logger.error(f"Failed to connect to Ollama: {e}")
        return False

iter_ndjson_chunks async 

iter_ndjson_chunks(chunk_iterator)

Async generator that yields parsed JSON objects from NDJSON (newline-delimited JSON) byte chunks.

Source code in src/ollama_mcp_bridge/utils.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def iter_ndjson_chunks(chunk_iterator):
    """Async generator that yields parsed JSON objects from NDJSON (newline-delimited JSON) byte chunks."""
    buffer = b""
    async for chunk in chunk_iterator:
        buffer += chunk
        while b"\n" in buffer:
            line, buffer = buffer.split(b"\n", 1)
            if line.strip():
                try:
                    yield json.loads(line)
                except json.JSONDecodeError as e:
                    logger.debug(f"Error parsing NDJSON line: {e}")
    # Handle any trailing data
    if buffer.strip():
        try:
            yield json.loads(buffer)
        except json.JSONDecodeError as e:
            logger.debug(f"Error parsing trailing NDJSON: {e}")

run_until_client_disconnects async 

run_until_client_disconnects(
    coro, request, poll_interval: float = 0.25
)

Run coro, cancelling it if the client disconnects first.

Nothing cancels a buffered handler when the client hangs up, so the bridge would keep driving Ollama for a client that left. Cancelling closes the connection to Ollama, which stops the generation there too.

The watcher reads from the ASGI receive channel, so the request body must already have been consumed or a body message could be swallowed.

Returns the result of coro, or raises ClientDisconnected.

Source code in src/ollama_mcp_bridge/utils.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def run_until_client_disconnects(coro, request, poll_interval: float = 0.25):
    """Run ``coro``, cancelling it if the client disconnects first.

    Nothing cancels a buffered handler when the client hangs up, so the bridge would keep
    driving Ollama for a client that left. Cancelling closes the connection to Ollama,
    which stops the generation there too.

    The watcher reads from the ASGI receive channel, so the request body must already have
    been consumed or a body message could be swallowed.

    Returns the result of ``coro``, or raises ``ClientDisconnected``.
    """
    task = asyncio.ensure_future(coro)
    watcher = asyncio.ensure_future(_wait_for_disconnect(request, poll_interval))
    try:
        done, _ = await asyncio.wait({task, watcher}, return_when=asyncio.FIRST_COMPLETED)
        if task in done:
            return task.result()
        watcher.result()  # a watcher that failed must not be read as a disconnect
        raise ClientDisconnected()
    finally:
        # Only the work is awaited: cancelling it is what closes the upstream connection.
        # Awaiting the watcher could hang, since anyio 3 (which starlette still allows)
        # swallows the cancel inside is_disconnected().
        watcher.cancel()
        if not task.done():
            task.cancel()
            await asyncio.gather(task, return_exceptions=True)

parse_upstream_headers 

parse_upstream_headers(
    env_value: Optional[str],
    header_flags: Optional[list] = None,
) -> Optional[Dict[str, str]]

Build the upstream headers dict from the UPSTREAM_HEADERS env var and CLI flags.

Parameters:

Name Type Description Default
env_value Optional[str]

Raw UPSTREAM_HEADERS env var, expected to be a JSON object of header name/value pairs (e.g. '{"Authorization": "Bearer xxx"}').

 required
header_flags Optional[list]

Repeatable --upstream-header values, each "Name: Value".

 None

CLI flags override env entries with the same header name. Returns None when no headers are configured.

Source code in src/ollama_mcp_bridge/utils.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def parse_upstream_headers(env_value: Optional[str], header_flags: Optional[list] = None) -> Optional[Dict[str, str]]:
    """Build the upstream headers dict from the UPSTREAM_HEADERS env var and CLI flags.

    Args:
        env_value: Raw UPSTREAM_HEADERS env var, expected to be a JSON object of
            header name/value pairs (e.g. '{"Authorization": "Bearer xxx"}').
        header_flags: Repeatable --upstream-header values, each "Name: Value".

    CLI flags override env entries with the same header name. Returns None when no
    headers are configured.
    """
    headers: Dict[str, str] = {}

    if env_value and env_value.strip():
        try:
            parsed = json.loads(env_value)
        except json.JSONDecodeError as e:
            raise BadParameter(f"UPSTREAM_HEADERS must be a valid JSON object: {e}")
        if not isinstance(parsed, dict):
            raise BadParameter("UPSTREAM_HEADERS must be a JSON object of header name/value pairs")
        for name, value in parsed.items():
            if not str(name).strip():
                raise BadParameter("UPSTREAM_HEADERS contains an empty header name")
            headers[str(name).strip()] = str(value)

    for raw in header_flags or []:
        name, sep, value = raw.partition(":")
        if not sep or not name.strip():
            raise BadParameter(f"Invalid --upstream-header {raw!r}, expected 'Name: Value'")
        headers[name.strip()] = value.strip()

    return headers or None

validate_cli_inputs 

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.

Parameters:

Name Type Description Default
system_prompt str

optional system prompt string; if provided, must be a non-empty string and not excessively long.

 None
Source code in src/ollama_mcp_bridge/utils.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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}")

    # Validate port
    if not 1 <= port <= 65535:
        raise BadParameter(f"Port must be between 1 and 65535, got {port}")

    # Validate host (basic check)
    if not isinstance(host, str) or not host:
        raise BadParameter("Host must be a non-empty string")

    # Validate URL (basic check)
    url_pattern = re.compile(r"^https?://[\w\.-]+(:\d+)?")
    if not url_pattern.match(ollama_url):
        raise BadParameter(f"Invalid Ollama URL: {ollama_url}")

    # Validate max_tool_rounds
    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)")

check_for_updates async 

check_for_updates(
    current_version: str, print_message: bool = False
) -> str

Check if a newer version of ollama-mcp-bridge is available on PyPI.

Parameters:

Name Type Description Default
current_version str

The current version of the package

 required
print_message bool

If True, print the update message to stdout instead of logging

 False

Returns:

Name Type Description
str str

The latest version if an update is available, otherwise the current version
Source code in src/ollama_mcp_bridge/utils.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
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.

    Args:
        current_version: The current version of the package
        print_message: If True, print the update message to stdout instead of logging

    Returns:
        str: The latest version if an update is available, otherwise the current version
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get("https://pypi.org/pypi/ollama-mcp-bridge/json", timeout=5)

            if response.status_code == 200:
                data = response.json()
                latest_version = data.get("info", {}).get("version", "0.0.0")

                # Compare versions
                current_v = pkg_version.parse(current_version)
                latest_v = pkg_version.parse(latest_version)

                if latest_v > current_v:
                    upgrade_cmd = "pip install --upgrade ollama-mcp-bridge"

                    # Show message based on requested output method
                    update_msg = f"📦 Update available: v{current_version} → v{latest_version}"
                    upgrade_msg = f"To upgrade, run: {upgrade_cmd}"

                    if print_message:
                        typer.echo(typer.style(update_msg, fg=typer.colors.BRIGHT_GREEN, bold=True))
                        typer.echo(typer.style(upgrade_msg, fg=typer.colors.BRIGHT_MAGENTA, bold=True))
                    else:
                        logger.info(update_msg)
                        logger.info(upgrade_msg)

                return latest_version

            return current_version  # Return current version when response doesn't match expected structure
    except (httpx.HTTPError, json.JSONDecodeError, pkg_version.InvalidVersion) as e:
        logger.debug(f"Failed to check for updates: {e}")
        return current_version  # Return current version when check fails

expand_env_vars 

expand_env_vars(value: str, cwd: str = None) -> str

Expand environment variable references in a string. Supports ${env:VAR_NAME} and ${workspaceFolder} syntax.

Source code in src/ollama_mcp_bridge/utils.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def expand_env_vars(value: str, cwd: str = None) -> str:
    """
    Expand environment variable references in a string.
    Supports ${env:VAR_NAME} and ${workspaceFolder} syntax.
    """
    if not isinstance(value, str):
        return value

    if cwd is None:
        cwd = os.getcwd()

    # Replace ${workspaceFolder} with current working directory
    value = value.replace("${workspaceFolder}", cwd)

    # Replace ${env:VAR_NAME} with environment variable value
    pattern = r"\$\{env:([^}]+)\}"
    matches = re.findall(pattern, value)
    for var_name in matches:
        env_value = os.getenv(var_name, "")
        value = value.replace(f"${{env:{var_name}}}", env_value)

    return value

expand_dict_env_vars 

expand_dict_env_vars(
    data: Dict[str, Any], cwd: str = None
) -> Dict[str, Any]

Recursively expand environment variables in a dictionary.

Source code in src/ollama_mcp_bridge/utils.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
def expand_dict_env_vars(data: Dict[str, Any], cwd: str = None) -> Dict[str, Any]:
    """Recursively expand environment variables in a dictionary."""
    result = {}
    for key, value in data.items():
        if isinstance(value, str):
            result[key] = expand_env_vars(value, cwd)
        elif isinstance(value, dict):
            result[key] = expand_dict_env_vars(value, cwd)
        elif isinstance(value, list):
            result[key] = [expand_env_vars(v, cwd) if isinstance(v, str) else v for v in value]
        else:
            result[key] = value
    return result