Skip to main content

​
Overview

The Ollama Python client integrates local language models with MCP tools, enabling LLMs to autonomously execute server-side functions. This creates a powerful agentic system for building AI applications with tool-calling capabilities.

​
Source Code

The implementation can be found at ~/workspace/source/clients/ollama-py/.

​
Prerequisites

  • Python 3.11 or higher
  • Ollama installed and running
  • A compatible LLM model (e.g., mistral:latest)
  • An MCP server

​
Installation

# Install uv
pip install uv

# Install dependencies
uv add requests mcp[cli] python-dotenv

​
Using pip

pip install requests "mcp[cli]>=1.5.0" python-dotenv

​
Architecture

The implementation consists of four main components:
  1. MCPClient: Manages MCP server connections and tool execution
  2. OllamaAPIClient: Handles Ollama API communication
  3. ToolManager: Converts MCP tools to Ollama format
  4. OllamaAgent: Orchestrates LLM and tool interactions

​
MCP Client Implementation

​
Class Structure

import logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from typing import Optional, Dict, Any

logger = logging.getLogger(__name__)

class MCPClient:
    """Client for interacting with MCP servers."""
    
    def __init__(self, command: str, args: list[str], env: Optional[Dict[str, str]] = None):
        self.server_params = StdioServerParameters(
            command=command,
            args=args,
            env=env
        )
        self.session = None
        self._client_ctx = None
        self._session_ctx = None

​
Connecting to MCP Server

async def connect(self) -> bool:
    """Establishes connection with the MCP server."""
    try:
        self._client_ctx = stdio_client(self.server_params)
        client = await self._client_ctx.__aenter__()
        self.read, self.write = client
        
        self._session_ctx = ClientSession(self.read, self.write)
        self.session = await self._session_ctx.__aenter__()
        await self.session.initialize()
        
        logger.info("Conexión exitosa con servidor MCP")
        return True
    except Exception as e:
        logger.error(f"Error al conectar: {e}")
        await self.disconnect()
        return False

​
Tool Operations

async def list_tools(self) -> Any:
    """Lists all available tools from the server."""
    if not self.session:
        raise RuntimeError("Cliente no conectado. Llama a connect() primero")
    
    tools = await self.session.list_tools()
    logger.debug(f"Herramientas disponibles: {tools}")
    return tools

async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
    """Executes a specific tool with arguments."""
    if not self.session:
        raise RuntimeError("Cliente no conectado. Llama a connect() primero")
    
    logger.debug(f"Ejecutando herramienta {tool_name} con argumentos: {arguments}")
    result = await self.session.call_tool(tool_name, arguments)
    logger.debug(f"Resultado: {result}")
    return result

​
Context Manager Support

async def __aenter__(self) -> 'MCPClient':
    """Async context manager entry."""
    success = await self.connect()
    if not success:
        raise RuntimeError("Error al conectar con el servidor MCP")
    return self

async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
    """Async context manager exit."""
    await self.disconnect()

​
Ollama API Client

​
Client Implementation

import requests
import json
from typing import List, Dict, Any, Union, Optional

class OllamaAPIClient:
    """Client for communicating with Ollama API."""
    
    def __init__(self, base_url: str = "http://localhost:11434"):
        self.base_url = base_url

    def check_connection(self) -> bool:
        """Verifies connection to Ollama."""
        try:
            response = requests.get(f"{self.base_url}/api/tags")
            if response.status_code != 200:
                raise Exception(f"Error al conectarse: {response.status_code}")
            return True
        except requests.exceptions.ConnectionError:
            raise Exception("No se pudo conectar al servidor de Ollama")

​
Chat with Tools

def chat(
    self, 
    model: str, 
    messages: List[Dict[str, Any]], 
    tools: List[Dict[str, Any]] = None,
    options: Optional[Dict[str, Any]] = None
) -> Union[str, Dict[str, Any]]:
    """Sends a chat request to Ollama with optional tools."""
    data = {
        "model": model,
        "messages": messages,
        "stream": False
    }
    
    if tools:
        data["tools"] = tools
        
    if options:
        data.update(options)
    
    response = requests.post(
        f"{self.base_url}/api/chat", 
        json=data, 
        timeout=60
    )
    
    if response.status_code != 200:
        logger.error(f"Error en la conversación: {response.status_code}")
        return None

    return self._process_response(response.text)

​
Response Processing

def _process_response(self, response_text: str) -> Union[str, Dict[str, Any]]:
    """Processes Ollama API response text."""
    lines = response_text.strip().split('\n')
    full_response = ""
    
    for line in lines:
        try:
            resp_json = json.loads(line)
            
            # Check for function call
            if "message" in resp_json and "tool_calls" in resp_json["message"]:
                function_call = resp_json["message"]["tool_calls"][0]
                if function_call:
                    return {
                        "type": "function_call",
                        "function_call": function_call
                    }
            
            # Accumulate normal response
            if "message" in resp_json and "content" in resp_json["message"]:
                content = resp_json["message"].get("content")
                if content:
                    full_response += content
                    
        except json.JSONDecodeError:
            logger.error(f"Error al decodificar: {line}")
            continue
    
    return full_response

​
Tool Manager

class ToolManager:
    """Manages tool conversion between MCP and Ollama formats."""
    
    def __init__(self):
        self.built_in_tools = []
    
    def get_all_tools(self, mcp_tools=None) -> List[Dict[str, Any]]:
        """Gets all available tools (built-in + MCP)."""
        tools = self.built_in_tools.copy()
        
        # Add MCP tools
        if mcp_tools and hasattr(mcp_tools, 'tools'):
            for mcp_tool in mcp_tools.tools:
                tools.append({
                    'type': 'function',
                    'function': {
                        'name': f"mcp_{mcp_tool.name}",
                        'description': getattr(mcp_tool, 'description', f"MCP tool: {mcp_tool.name}"),
                        'parameters': getattr(mcp_tool, 'inputSchema', {'type': 'object'})
                    }
                })
                
        return tools

​
Ollama Agent

​
Agent Class

from enum import Enum

class MessageRole(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"

class OllamaAgent:
    """Agent integrating Ollama with MCP tools."""
    
    def __init__(
        self, 
        ollama_url: str = "http://localhost:11434",
        mcp_command: str = "node",
        mcp_args: List[str] = None
    ):
        if mcp_args is None:
            mcp_args = ["/path/to/server.js"]
            
        self.ollama_client = OllamaAPIClient(ollama_url)
        self.mcp_client = MCPClient(mcp_command, mcp_args)
        self.tool_manager = ToolManager()
        self.toolsMCP = None
        
        # Verify Ollama connection
        self.ollama_client.check_connection()
        logger.info("✅ Conexión establecida con Ollama")

​
Setup and Context Manager

async def __aenter__(self):
    """Async context manager entry."""
    await self.setup()
    return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Async context manager exit."""
    if self.mcp_client:
        await self.mcp_client.__aexit__(exc_type, exc_val, exc_tb)

async def setup(self):
    """Sets up the agent and MCP tools."""
    try:
        if self.mcp_client:
            await self.mcp_client.__aenter__()
            self.toolsMCP = await self.mcp_client.list_tools()
            logger.info("✅ Conexión establecida con servidor MCP")
    except Exception as e:
        logger.error(f"❌ Error al conectarse al servidor MCP: {e}")
        self.toolsMCP = None

​
Tool Execution

async def execute_mcp_tool(self, tool_name: str, arguments: Dict[str, Any]):
    """Executes an MCP tool."""
    if not self.mcp_client or not self.toolsMCP:
        raise RuntimeError("MCP Client not connected or tools not available")
    
    try:
        return await self.mcp_client.execute_tool(tool_name, arguments)
    except Exception as e:
        logger.error(f"Error executing MCP tool {tool_name}: {e}")
        raise

def chat(self, model: str, messages: List[Dict[str, Any]], options: Optional[Dict[str, Any]] = None):
    """Conducts a conversation with the model."""
    tools = self.tool_manager.get_all_tools(self.toolsMCP)
    return self.ollama_client.chat(model, messages, tools, options)

​
Function Execution Flow

​
Execute Function Handler

async def execute_function(function_name: str, function_args: dict, agent: OllamaAgent) -> str:
    """Executes a function with its arguments."""
    try:
        # Check if it's an MCP tool
        if function_name.startswith("mcp_"):
            actual_tool_name = function_name[4:]
            try:
                logger.info(f"Ejecutando herramienta MCP: {actual_tool_name}")
                result = await agent.execute_mcp_tool(actual_tool_name, function_args)
                logger.info(f"Resultado: {result}")
                return str(result)
            except Exception as e:
                return f"Error ejecutando la herramienta MCP {actual_tool_name}: {e}"
        
        return f"Función {function_name} no implementada"
    except Exception as e:
        logger.error(f"Error ejecutando la función {function_name}: {e}")
        import traceback
        traceback.print_exc()
        return f"Error ejecutando la función: {e}"

​
Processing Function Calls

async def process_function_call(model_name: str, response: dict, messages: list, agent: OllamaAgent):
    """Processes a function call from the model."""
    try:
        function_call = response["function_call"]
        function_name = function_call["function"]["name"]
        
        # Parse arguments
        function_args_str = function_call["function"]["arguments"]
        function_args = function_args_str if isinstance(function_args_str, dict) else json.loads(function_args_str)
        
        function_call_id = "call_" + str(len(messages))
        
        logger.info(f"\n{model_name} quiere llamar a la función: {function_name}")
        logger.info(f"Con los argumentos: {json.dumps(function_args, indent=2)}")
        
        # Execute function
        function_result = await execute_function(function_name, function_args, agent)
        
        logger.info(f"Resultado: {function_result}")
        
        # Add to message history
        messages.append({
            "role": MessageRole.ASSISTANT, 
            "content": None,
            "tool_calls": [{
                "id": function_call_id,
                "function": {
                    "name": function_name,
                    "arguments": function_args_str
                }
            }]
        })
        
        messages.append({
            "role": MessageRole.TOOL,
            "tool_call_id": function_call_id,
            "name": function_name,
            "content": function_result
        })
        
        # Get final response
        final_response = agent.chat(model_name, messages)
        
        if isinstance(final_response, dict) and final_response.get("type") == "function_call":
            await process_function_call(model_name, final_response, messages, agent)
        elif isinstance(final_response, str):
            print(f"\n{model_name}: {final_response}")
            messages.append({"role": MessageRole.ASSISTANT, "content": final_response})
    except Exception as e:
        logger.error(f"Error procesando la llamada a función: {e}")
        import traceback
        traceback.print_exc()

​
Interactive Chat

async def interactive_chat(agent: OllamaAgent):
    """Interactive chat mode with Ollama."""
    model_name = "mistral:latest"
    
    # Verify model exists
    models = agent.list_models()
    model_exists = any(model["name"] == model_name for model in models)
    
    if not model_exists:
        logger.warning(f"El modelo '{model_name}' no está disponible")
        if models:
            model_name = models[0]["name"]
            logger.info(f"Usando el modelo: {model_name}")
        else:
            logger.error("No hay modelos disponibles")
            return
    
    # Start chat
    messages = []
    messages.append({
        "role": MessageRole.SYSTEM, 
        "content": "Eres un agente que consultará las tools disponibles",
    })
    
    print("\nIniciando chat (escribe '/salir' para terminar)")
    
    while True:
        try:
            user_message = input("\nTú: ")
            
            if user_message.lower() in ["/salir", "/exit", "/quit"]:
                break
            
            messages.append({"role": MessageRole.USER, "content": user_message})
            
            print("Generando respuesta...")
            response = agent.chat(model_name, messages)
            
            if response:
                if isinstance(response, dict) and response.get("type") == "function_call":
                    await process_function_call(model_name, response, messages, agent)
                elif isinstance(response, str):
                    print(f"\n{model_name}: {response}")
                    messages.append({"role": MessageRole.ASSISTANT, "content": response})
        except KeyboardInterrupt:
            print("\nChat interrumpido")
            break
        except Exception as e:
            logger.error(f"Error en el chat: {e}")

​
Main Application

import asyncio

async def main():
    """Main function."""
    async with OllamaAgent() as agent:
        await interactive_chat(agent)

if __name__ == "__main__":
    # Configure logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    asyncio.run(main())

​
Running the Application

​
Set Environment Variables

export MCP_SERVER_PATH="/path/to/your/server.js"

​
Run the Application

python ollama-python-app.py

​
Configuration

DEFAULT_OLLAMA_URL = "http://localhost:11434"
DEFAULT_MCP_SERVER_COMMAND = "node"
DEFAULT_MCP_SERVER_PATH = os.environ.get(
    "MCP_SERVER_PATH", 
    "/path/to/server.js"
)
DEFAULT_MODEL = "mistral:latest"

​
Best Practices

  1. Use Context Managers: Always use async with for proper cleanup
  2. Error Handling: Wrap tool calls in try-except blocks
  3. Logging: Use the logging module instead of print statements
  4. Type Hints: Include type hints for better IDE support
  5. Model Selection: Use models with function calling support

​
Troubleshooting

​
Ollama Connection Issues

# Check Ollama status
ollama list

# Start Ollama
ollama serve

# Pull a model
ollama pull mistral:latest

​
MCP Connection Errors

  • Verify server path in environment variable
  • Check that Node.js is installed
  • Ensure MCP server is properly compiled

​
Function Calling Not Working

  • Use models that support function calling (Mistral, Llama 3 70B)
  • Check tool definitions are properly formatted
  • Verify message history structure

​
Next Steps

Build docs developers (and LLMs) love

Get started for freeTalk to us