Construire des Agents IA avec DeepSeek et le Model Context Protocol (MCP) : Un Guide Pratique Utilisant le SDK Agentic d'OpenAI

python -m venv env
source env/bin/activate
pip install openai-agents

python agent_mcp_deepseek.py --repo_path /chemin/vers/votre/depot

#!/usr/bin/env python3

"""
This script runs a Model Context Protocol (MCP) client that interacts with MCP servers
to answer questions about a specified git repository and its filesystem.

It utilizes an AI model (via OpenAI API or compatible services like DeepSeek)
to generate responses based on the context provided by the MCP servers.

Usage:
    python agent_mcp_deepseek.py --repo_path <path_to_your_git_repo>

Prerequisites:
- Python 3.7+
- Required Python packages (install via pip, e.g., openai, agents)
- `uvx` tool installed (`pip install uvx`)
- `npx` tool installed (comes with Node.js/npm)
- Environment variable `DEEPSEEK_API_KEY` set with your API key.
"""

import os
import asyncio
import shutil
import argparse
import logging
from typing import List

from openai import AsyncOpenAI

from agents import Agent, Runner, set_tracing_disabled, OpenAIChatCompletionsModel
from agents.mcp import MCPServer, MCPServerStdio

# --- Configuration ---

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# AI Model Configuration (DeepSeek API or compatible)
BASE_URL = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")
API_KEY = os.getenv("DEEPSEEK_API_KEY")
MODEL_NAME = os.getenv("DEEPSEEK_MODEL_NAME", "deepseek-chat")

# ANSI escape codes for colored console output
DARK_BLUE = "\033[34m"
RESET = "\033[0m"

# --- Pre-run Checks ---

# Validate essential configuration
if not API_KEY:
    raise ValueError(
        "API key is missing. Please set the DEEPSEEK_API_KEY environment variable."
    )
if not BASE_URL:
    logging.warning("DEEPSEEK_BASE_URL not set, using default: %s", BASE_URL)
if not MODEL_NAME:
    logging.warning("DEEPSEEK_MODEL_NAME not set, using default: %s", MODEL_NAME)

# Check for required external tools
if not shutil.which("uvx"):
    raise RuntimeError("uvx command not found. Please install it, e.g., `pip install uvx`.")
if not shutil.which("npx"):
    raise RuntimeError("npx command not found. Please ensure Node.js and npm/npx are installed and in your system's PATH.")

# --- Core Logic ---

async def run_agent_query(
    mcp_servers: List[MCPServer],
    directory_path: str,
    user_query: str
) -> None:
    """Runs the AI agent to answer a specific query about the repository/filesystem.

    Args:
        mcp_servers: A list of active MCP server instances.
        directory_path: The path to the target git repository and filesystem root.
        user_query: The question asked by the user.
    """
    logging.info(f"Processing query for directory: {directory_path}")

    # Initialize the AI client
    client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)

    # Disable agent tracing for cleaner output in this interactive tool
    set_tracing_disabled(disabled=True)

    # Create the AI agent instance
    agent = Agent(
        name="Assistant",
        instructions=(
            f"You answer questions about the git repository and filesystem located at "
            f"'{directory_path}'. Use this path for context when interacting with tools."
        ),
        mcp_servers=mcp_servers,
        model=OpenAIChatCompletionsModel(model=MODEL_NAME, openai_client=client),
    )

    # Run the agent with the user's query
    print("\n" + "-" * 40)
    print(f"{DARK_BLUE}Question: {user_query}{RESET}")
    print("Assistant thinking...")
    try:
        result = await Runner.run(starting_agent=agent, input=user_query)
        print(f"{DARK_BLUE}Answer:\n{result.final_output}{RESET}")
    except Exception as e:
        logging.error(f"Error during agent execution: {e}", exc_info=True)
        print(f"Sorry, an error occurred while processing your request: {e}")
    finally:
        print("-" * 40 + "\n")


async def main() -> None:
    """Sets up MCP servers and handles the interactive user query loop.

    Parses command-line arguments for the repository path, starts the Git and
    Filesystem MCP servers, and then enters a loop to accept user questions,
    running the agent for each query until the user quits.
    """
    parser = argparse.ArgumentParser(
        description="Run an AI assistant powered by Model Context Protocol (MCP) servers to query a git repository."
    )
    parser.add_argument(
        "--repo_path",
        type=str,
        required=True,
        help="Path to the target git repository."
    )
    args = parser.parse_args()

    directory_path = os.path.abspath(args.repo_path) # Use absolute path
    logging.info(f"Target repository path: {directory_path}")

    if not os.path.isdir(directory_path):
        raise FileNotFoundError(f"The specified repository path does not exist or is not a directory: {directory_path}")

    # Start MCP servers concurrently
    try:
        async with MCPServerStdio(
            name="Git Server (via uvx)",
            cache_tools_list=True, # Example: Cache tool list for performance
            params={"command": "uvx", "args": ["mcp-server-git"]},
            # Note: mcp-server-git typically doesn't need a path argument,
            # it operates on the repo where it's invoked or uses tool parameters.
        ) as git_server, MCPServerStdio(
            name="Filesystem Server (via npx)",
            params={
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", directory_path],
            },
        ) as fs_server:
            logging.info("Model Context Protocol (MCP) servers (Git, Filesystem) started successfully.")
            servers = [git_server, fs_server]

            # Interactive query loop
            while True:
                try:
                    message = input("Enter your question (or type '/quit' to exit): ")
                except EOFError:
                    print("\nExiting.") # Handle Ctrl+D
                    break

                if message.strip().lower() == "/quit":
                    logging.info("'/quit' command received. Exiting.")
                    break
                if not message.strip():
                    continue # Ignore empty input

                logging.info(f"User question: {message}")
                await run_agent_query(servers, directory_path, message)

    except Exception as e:
        logging.critical(f"Failed to start or run Model Context Protocol (MCP) servers: {e}", exc_info=True)
        print(f"Critical error: {e}. Exiting.")
    finally:
        logging.info("Model Context Protocol (MCP) toolkit finished.")


if __name__ == "__main__":
    logging.info("Starting Model Context Protocol (MCP) toolkit...")
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nInterrupted by user. Exiting.")
        logging.info("Process interrupted by user (KeyboardInterrupt).")
    except (ValueError, RuntimeError, FileNotFoundError) as e:
        logging.error(f"Initialization failed: {e}")
        print(f"Error: {e}")
        # Exit with error code if setup fails
        exit(1)