بناء خوادم MCP للإنتاج باستخدام FastMCP في Python

uv init project-mcp-server
cd project-mcp-server
uv add fastmcp

mkdir project-mcp-server && cd project-mcp-server
python -m venv .venv && source .venv/bin/activate
pip install fastmcp

fastmcp version
# FastMCP 3.2.4

touch server.py

from fastmcp import FastMCP, Context
 
mcp = FastMCP(
    name="Project Intelligence Server",
    instructions=(
        "لديك صلاحية الوصول إلى قاعدة بيانات المشاريع. "
        "استخدم search_projects للبحث بالكلمات المفتاحية، "
        "وget_project_status لاسترجاع البيانات الحية."
    ),
)

PROJECTS = [
    {"id": "p1", "name": "منصة نقطة", "status": "active", "lang": "TypeScript"},
    {"id": "p2", "name": "API الفواتير", "status": "active", "lang": "Python"},
    {"id": "p3", "name": "تطبيق الجوال", "status": "paused", "lang": "React Native"},
    {"id": "p4", "name": "مسار البيانات", "status": "archived", "lang": "Python"},
]
 
 
@mcp.tool
def search_projects(keyword: str, status: str = "all") -> list[dict]:
    """
    البحث في المشاريع بكلمة مفتاحية وفلتر اختياري للحالة.
 
    Args:
        keyword: مصطلح البحث يُطابق اسم المشروع واللغة.
        status: تصفية حسب الحالة: active أو paused أو archived أو all.
    """
    results = PROJECTS
    if status != "all":
        results = [p for p in results if p["status"] == status]
    keyword = keyword.lower()
    return [p for p in results if keyword in p["name"].lower() or keyword in p["lang"].lower()]
 
 
@mcp.tool
async def get_project_status(project_id: str, ctx: Context) -> dict:
    """
    استرجاع الحالة التفصيلية لمشروع محدد برقمه التعريفي.
 
    Args:
        project_id: المعرّف الفريد للمشروع (مثلاً p1 أو p2).
    """
    await ctx.info(f"جارٍ جلب بيانات المشروع {project_id}")
 
    project = next((p for p in PROJECTS if p["id"] == project_id), None)
    if project is None:
        raise ValueError(f"المشروع '{project_id}' غير موجود")
 
    return {
        **project,
        "last_commit": "2026-06-24",
        "open_issues": 3,
        "coverage": "87%",
    }

import json
 
 
@mcp.resource("projects://list")
def list_all_projects() -> str:
    """لقطة من جميع المشاريع في النظام."""
    return json.dumps(PROJECTS, ensure_ascii=False, indent=2)
 
 
@mcp.resource("projects://{project_id}/details")
def project_details(project_id: str) -> str:
    """عرض تفصيلي لمشروع واحد عبر معرّفه."""
    project = next((p for p in PROJECTS if p["id"] == project_id), None)
    if project is None:
        return json.dumps({"error": f"المشروع {project_id} غير موجود"}, ensure_ascii=False)
    return json.dumps(project, ensure_ascii=False, indent=2)

@mcp.prompt
def status_report(project_id: str, audience: str = "engineering") -> str:
    """
    إنشاء قالب تقرير حالة لمشروع محدد وجمهور مستهدف.
 
    Args:
        project_id: معرّف المشروع المستهدف.
        audience: الجمهور المستهدف: engineering أو management أو client.
    """
    tone = {
        "engineering": "تقني، يركّز على المقاييس والعوائق",
        "management": "موجز، يركّز على الجدول الزمني والمخاطر",
        "client": "غير تقني، إيجابي، يركّز على القيمة المُسلَّمة",
    }.get(audience, "متوازن")
 
    return (
        f"باستخدام بيانات المشروع '{project_id}'، اكتب تقرير حالة {tone}. "
        "اشمل: الحالة الحالية، التقدم الأخير، المشكلات المفتوحة، والمعالم القادمة. "
        "لا يتجاوز 300 كلمة."
    )

from contextlib import asynccontextmanager
from fastmcp.dependencies import Depends
 
 
async def get_db():
    """محاكاة عميل قاعدة البيانات. في الإنتاج، استبدله باتصال حقيقي."""
    class FakeDB:
        async def query(self, sql: str) -> list:
            return [{"result": f"نُفِّذ: {sql}"}]
 
    db = FakeDB()
    yield db
 
 
@mcp.tool
async def run_query(sql: str, db=Depends(get_db)) -> list:
    """
    تنفيذ استعلام SQL للقراءة فقط على قاعدة بيانات المشاريع.
 
    Args:
        sql: جملة SELECT للتنفيذ.
    """
    return await db.query(sql)

from fastmcp.server.lifespan import lifespan
 
 
@lifespan
async def app_lifespan(server):
    """تهيئة وتنظيف الموارد على مستوى الخادم."""
    print("الخادم يبدأ — جارٍ بناء فهرس المشاريع...")
    index = {p["id"]: p for p in PROJECTS}
    yield {"index": index}
    print("الخادم يُغلَق — تحرير الفهرس.")
 
 
mcp_with_lifespan = FastMCP(
    name="Project Intelligence Server",
    lifespan=app_lifespan,
)
 
 
@mcp_with_lifespan.tool
def fast_lookup(project_id: str, ctx: Context) -> dict:
    """بحث فوري باستخدام الفهرس المُبنى مسبقاً."""
    index = ctx.lifespan_context["index"]
    return index.get(project_id, {"error": "غير موجود"})

fastmcp dev inspector server.py

{
  "mcpServers": {
    "project-intelligence": {
      "command": "fastmcp",
      "args": ["run", "/المسار/الكامل/إلى/server.py"]
    }
  }
}

claude mcp add project-intelligence fastmcp run /المسار/الكامل/إلى/server.py

{
  "name": "project-intelligence",
  "command": "fastmcp run /المسار/الكامل/إلى/server.py"
}

from fastmcp.server.auth import StaticTokenVerifier
import os
 
auth = StaticTokenVerifier(
    tokens={
        os.environ["MCP_TOKEN_PROD"]: {"client_id": "prod", "scopes": ["read", "write"]},
        os.environ["MCP_TOKEN_READONLY"]: {"client_id": "viewer", "scopes": ["read"]},
    },
    required_scopes=["read"],
)
 
secured_mcp = FastMCP(
    name="Project Intelligence Server",
    auth=auth,
)
 
if __name__ == "__main__":
    secured_mcp.run(transport="http", host="0.0.0.0", port=8000)

MCP_TOKEN_PROD=your-secret-token \
MCP_TOKEN_READONLY=your-readonly-token \
python server.py

FROM python:3.12-slim
 
WORKDIR /app
 
RUN pip install uv
 
COPY pyproject.toml uv.lock* ./
RUN uv sync --frozen --no-dev
 
COPY server.py .
 
EXPOSE 8000
 
CMD ["uv", "run", "fastmcp", "run", "server.py", \
     "--transport", "http", "--host", "0.0.0.0", "--port", "8000"]

docker build -t project-mcp-server .
docker run -p 8000:8000 \
  -e MCP_TOKEN_PROD=your-secret-token \
  project-mcp-server

from fastmcp.telemetry import get_tracer
 
 
@mcp.tool
async def search_projects_traced(keyword: str, ctx: Context) -> list[dict]:
    """بحث في المشاريع مع تتبع موزّع."""
    tracer = get_tracer()
 
    with tracer.start_as_current_span("search.filter") as span:
        span.set_attribute("search.keyword", keyword)
        results = [p for p in PROJECTS if keyword.lower() in p["name"].lower()]
        span.set_attribute("search.result_count", len(results))
 
    await ctx.info(f"وُجد {len(results)} مشروع للكلمة '{keyword}'")
    return results

OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317 \
OTEL_SERVICE_NAME=project-mcp-server \
python server.py