Πρακτικός Οδηγός Anthropic MCP: Πώς να Στήσεις Δικό σου MCP Server

# Δημιουργία φακέλου εργασίας
mkdir custom-mcp-server
cd custom-mcp-server

# Εγκατάσταση του επίσημου SDK της Anthropic και απαραίτητων βιβλιοθηκών
pip install "mcp[cli]" httpx pydantic

# Αν χρησιμοποιείς το ταχύτατο εργαλείο uv:
# uv venv
# uv add "mcp[cli]" httpx pydantic

import os
import sqlite3
from typing import Any, Dict, List, Optional
import httpx
from mcp.server.fastmcp import FastMCP, Context

# 1. Αρχικοποίηση του FastMCP Server
mcp = FastMCP(
    name="DarkMonster-LocalHub",
    dependencies=["httpx", "pydantic"]
)

# Ρυθμίσεις περιβάλλοντος
DB_PATH = os.getenv("SQLITE_DB_PATH", "company_vault.db")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "")

# Βοηθητική μέθοδος για ασφαλή σύνδεση στη βάση (Read-Only από προεπιλογή)
def get_db_connection(readonly: bool = True):
    if not os.path.exists(DB_PATH):
        # Δημιουργία δοκιμαστικής βάσης αν δεν υπάρχει
        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()
        cur.execute("""
            CREATE TABLE IF NOT EXISTS projects (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                status TEXT NOT NULL,
                budget REAL
            )
        """)
        cur.execute("INSERT OR IGNORE INTO projects VALUES (1, 'Project Cerberus', 'In Progress', 45000.0)")
        cur.execute("INSERT OR IGNORE INTO projects VALUES (2, 'Quantum Sentinel', 'Review', 120000.0)")
        conn.commit()
        conn.close()

    # Σύνδεση σε Read-Only URI mode για μέγιστη ασφάλεια
    uri = f"file:{os.path.abspath(DB_PATH)}?mode=ro" if readonly else os.path.abspath(DB_PATH)
    return sqlite3.connect(uri, uri=True)


# ==========================================
# ΕΡΓΑΛΕΙΟ 1: Εκτέλεση SQL Query (Read-Only)
# ==========================================
@mcp.tool()
def query_sqlite(sql_query: str) -> List[Dict[str, Any]]:
    """
    Εκτελεί ένα ασφαλές SELECT query στην τοπική βάση SQLite και επιστρέφει αποτελέσματα.
    Επιτρέπονται ΜΟΝΟ queries ανάγνωσης (SELECT). Οποιεσδήποτε εντολές INSERT, UPDATE,
    DELETE ή DROP απορρίπτονται αυτόματα για ασφάλεια.
    """
    normalized = sql_query.strip().lower()
    forbidden_words = ["drop", "delete", "truncate", "update", "insert", "alter", "vacuum"]
    if any(word in normalized.split() for word in forbidden_words):
        raise ValueError("Ασφάλεια: Επιτρέπονται μόνο SELECT queries στη βάση!")

    conn = get_db_connection(readonly=True)
    conn.row_factory = sqlite3.Row
    try:
        cur = conn.cursor()
        cur.execute(sql_query)
        rows = cur.fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()


# ==========================================
# ΕΡΓΑΛΕΙΟ 2: Λήψη Σχήματος Πινάκων (Schema)
# ==========================================
@mcp.tool()
def get_table_schema(table_name: str) -> List[Dict[str, Any]]:
    """
    Επιστρέφει τη δομή στηλών, τύπους δεδομένων και constraints για έναν συγκεκριμένο πίνακα SQLite.
    """
    conn = get_db_connection(readonly=True)
    try:
        cur = conn.cursor()
        cur.execute(f"PRAGMA table_info({table_name});")
        columns = cur.fetchall()
        return [{"cid": c[0], "name": c[1], "type": c[2], "notnull": c[3], "pk": c[5]} for c in columns]
    finally:
        conn.close()


# ==========================================
# RESOURCE: Αυτόματο Context για τη Βάση
# ==========================================
@mcp.resource("sqlite://schema")
def get_database_schema_resource() -> str:
    """Εκθέτει τη λίστα όλων των πινάκων και των DDL schemas ως αυτόματο Resource στο Claude."""
    conn = get_db_connection(readonly=True)
    try:
        cur = conn.cursor()
        cur.execute("SELECT name, sql FROM sqlite_master WHERE type='table';")
        tables = cur.fetchall()
        output = ["=== SQLITE DATABASE STRUCTURE ==="]
        for name, sql in tables:
            output.append(f"Table: {name}\nDDL: {sql}\n")
        return "\n".join(output)
    finally:
        conn.close()


# ==========================================
# ΕΡΓΑΛΕΙΟ 3: GitHub Issues & PRs Inspector
# ==========================================
@mcp.tool()
def search_github_issues(repo_owner: str, repo_name: str, state: str = "open", limit: int = 5) -> List[Dict[str, Any]]:
    """
    Αναζητά issues και pull requests σε ένα αποθετήριο του GitHub.
    Χρησιμοποιεί το GITHUB_TOKEN από το περιβάλλον αν έχει οριστεί.
    """
    headers = {"Accept": "application/vnd.github.v3+json", "User-Agent": "FastMCP-Agent"}
    if GITHUB_TOKEN:
        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"

    url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/issues"
    params = {"state": state, "per_page": min(limit, 20)}

    with httpx.Client(timeout=10.0) as client:
        res = client.get(url, headers=headers, params=params)
        if res.status_code != 200:
            return [{"error": f"GitHub API error {res.status_code}: {res.text}"}]
        
        data = res.json()
        issues_summary = []
        for item in data:
            issues_summary.append({
                "number": item.get("number"),
                "title": item.get("title"),
                "user": item.get("user", {}).get("login"),
                "state": item.get("state"),
                "created_at": item.get("created_at"),
                "html_url": item.get("html_url"),
                "comments": item.get("comments"),
                "is_pr": "pull_request" in item
            })
        return issues_summary


# ==========================================
# ΠΥΛΩΝΑΣ 3: MCP PROMPTS (Workflow Templates)
# ==========================================
@mcp.prompt()
def audit_project_and_issues(project_name: str, github_repo: str) -> str:
    """
    Δημιουργεί ένα εξειδικευμένο prompt template για το Claude Desktop, καθοδηγώντας
    τον agent να συσχετίσει το budget ενός τοπικού project με τα ανοιχτά issues στο GitHub.
    """
    return f"""Είσαι ένας Senior Technical Project Manager και Security Auditor.
Σου ζητείται να αναλύσεις το έργο '{project_name}' στην τοπική μας βάση SQLite και να το συγκρίνεις με τα issues του αποθετηρίου '{github_repo}' στο GitHub.

Ακολούθησε αυστηρά τα παρακάτω βήματα:
1. Κάλεσε το εργαλείο 'query_sqlite' για να ανακτήσεις το τρέχον status και budget του project '{project_name}'.
2. Κάλεσε το εργαλείο 'search_github_issues' για να ανακτήσεις τα ανοιχτά bugs και pull requests του '{github_repo}'.
3. Εξέτασε αν υπάρχουν κρίσιμα τεχνικά εμπόδια (blockers) που θέτουν σε κίνδυνο την υλοποίηση.
4. Παράδωσε μια οργανωμένη αναφορά με:
   - Σύνοψη κατάστασης (Executive Status)
   - Οικονομικό αποτύπωμα (Budget vs Open Risks)
   - Προτεινόμενες άμεσες ενέργειες (Next Best Actions)."""


# ==========================================
# ADVANCED CAPABILITY: CLIENT SAMPLING
# ==========================================
@mcp.tool()
async def smart_issue_triage(issue_title: str, issue_body: str, ctx: Context) -> str:
    """
    Επιδεικνύει το Client Sampling: Ο MCP Server ζητά από το LLM του Host (Claude)
    να εξάγει μια συνοπτική εκτίμηση κινδύνου, χωρίς ο server να χρειάζεται δικό του API key!
    """
    # Ο server ζητά completion απευθείας από το LLM του Claude Desktop
    sampling_response = await ctx.session.create_message(
        messages=[{
            "role": "user",
            "content": {
                "type": "text",
                "text": f"Κατηγοριοποίησε σε 1 πρόταση το παρακάτω bug (Severity: Low/Medium/Critical) και δώσε εκτίμηση impact:\nTitle: {issue_title}\nBody: {issue_body}"
            }
        }],
        max_tokens=120
    )
    return sampling_response.content.text


if __name__ == "__main__":
    # Εκκίνηση του server με stdio transport (για απευθείας σύνδεση στο Claude Desktop)
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "darkmonster-hub": {
      "command": "python",
      "args": [
        "C:\\Users\\prive\\custom-mcp-server\\server.py"
      ],
      "env": {
        "SQLITE_DB_PATH": "C:\\Users\\prive\\custom-mcp-server\\company_vault.db",
        "GITHUB_TOKEN": "ghp_yourActualGitHubTokenHere"
      }
    }
  }
}

# Εκτέλεση του Inspector μέσω npx (απαιτείται Node.js 18+)
npx @modelcontextprotocol/inspector python server.py