Kendi MCP Sunucunuzu Yazın

mkdir tasks-mcp && cd tasks-mcp
npm init -y
npm install @modelcontextprotocol/sdk
npm install -D typescript tsx @types/node
npx tsc --init

{
  "type": "module",
  "scripts": {
    "start": "tsx src/index.ts",
    "build": "tsc"
  }
}

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { readFile, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { homedir } from "node:os";
import { randomUUID } from "node:crypto";

const STORE = join(homedir(), ".tasks.json");

type Task = {
  id: string;
  title: string;
  due?: string;       // ISO date
  tags?: string[];
  done: boolean;
  created: string;
};

async function loadTasks(): Promise<Task[]> {
  try {
    const raw = await readFile(STORE, "utf8");
    return JSON.parse(raw);
  } catch {
    return [];
  }
}

async function saveTasks(tasks: Task[]) {
  await writeFile(STORE, JSON.stringify(tasks, null, 2), "utf8");
}

const server = new Server(
  { name: "tasks-mcp", version: "0.1.0" },
  { capabilities: { tools: {} } },
);

// 1. Hangi tool'ları sunduğumuzu söyle
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "list_tasks",
      description:
        "Yapılacaklar listesini döner. Tamamlanmış/açık filtresi ve son tarih filtresi opsiyoneldir.",
      inputSchema: {
        type: "object",
        properties: {
          status: { type: "string", enum: ["open", "done", "all"] },
          due_before: { type: "string", description: "ISO tarihi; bu tarihten önce due'su olanları getirir" },
        },
      },
    },
    {
      name: "add_task",
      description: "Yeni bir görev ekler. due isteğe bağlı; tags isteğe bağlı, virgülle ayrılmaz, dizi olur.",
      inputSchema: {
        type: "object",
        properties: {
          title: { type: "string", description: "Görev başlığı" },
          due: { type: "string", description: "Son tarih, ISO formatında" },
          tags: {
            type: "array",
            items: { type: "string" },
            description: "Etiket dizisi (örn. ['iş', 'rapor'])",
          },
        },
        required: ["title"],
      },
    },
    {
      name: "complete_task",
      description: "Bir görevi tamamlandı olarak işaretler.",
      inputSchema: {
        type: "object",
        properties: { id: { type: "string", description: "Görev ID'si" } },
        required: ["id"],
      },
    },
    {
      name: "search_tasks",
      description: "Başlığa veya etikete göre görev arar.",
      inputSchema: {
        type: "object",
        properties: { query: { type: "string" } },
        required: ["query"],
      },
    },
  ],
}));

// 2. Çağrıları işle
server.setRequestHandler(CallToolRequestSchema, async (req) => {
  const { name, arguments: args } = req.params as { name: string; arguments: any };
  const tasks = await loadTasks();

  switch (name) {
    case "list_tasks": {
      const status = args?.status ?? "open";
      const dueBefore = args?.due_before ? new Date(args.due_before) : null;
      const filtered = tasks.filter((t) => {
        if (status === "open" && t.done) return false;
        if (status === "done" && !t.done) return false;
        if (dueBefore && t.due && new Date(t.due) >= dueBefore) return false;
        return true;
      });
      return {
        content: [{ type: "text", text: JSON.stringify(filtered, null, 2) }],
      };
    }

    case "add_task": {
      const task: Task = {
        id: randomUUID().slice(0, 8),
        title: args.title,
        due: args.due,
        tags: args.tags,
        done: false,
        created: new Date().toISOString(),
      };
      tasks.push(task);
      await saveTasks(tasks);
      return {
        content: [{ type: "text", text: `Eklendi: ${task.id} — ${task.title}` }],
      };
    }

    case "complete_task": {
      const t = tasks.find((x) => x.id === args.id);
      if (!t) {
        return {
          content: [{ type: "text", text: `Görev bulunamadı: ${args.id}` }],
          isError: true,
        };
      }
      t.done = true;
      await saveTasks(tasks);
      return {
        content: [{ type: "text", text: `Tamamlandı: ${t.title}` }],
      };
    }

    case "search_tasks": {
      const q = (args.query as string).toLowerCase();
      const hits = tasks.filter(
        (t) =>
          t.title.toLowerCase().includes(q) ||
          (t.tags ?? []).some((tag) => tag.toLowerCase().includes(q)),
      );
      return {
        content: [{ type: "text", text: JSON.stringify(hits, null, 2) }],
      };
    }

    default:
      throw new Error(`Bilinmeyen tool: ${name}`);
  }
});

// 3. stdio üzerinden bağlan
const transport = new StdioServerTransport();
await server.connect(transport);

pip install mcp

import json
import os
import uuid
from datetime import datetime
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

STORE = Path.home() / ".tasks.json"


def load_tasks() -> list[dict]:
    if not STORE.exists():
        return []
    return json.loads(STORE.read_text("utf-8"))


def save_tasks(tasks: list[dict]) -> None:
    STORE.write_text(json.dumps(tasks, indent=2, ensure_ascii=False), "utf-8")


server = Server("tasks-mcp")


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="list_tasks",
            description="Yapılacaklar listesini döner.",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {"type": "string", "enum": ["open", "done", "all"]},
                    "due_before": {"type": "string"},
                },
            },
        ),
        Tool(
            name="add_task",
            description="Yeni bir görev ekler.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "due": {"type": "string"},
                    "tags": {"type": "array", "items": {"type": "string"}},
                },
                "required": ["title"],
            },
        ),
        Tool(
            name="complete_task",
            description="Görevi tamamlandı olarak işaretler.",
            inputSchema={
                "type": "object",
                "properties": {"id": {"type": "string"}},
                "required": ["id"],
            },
        ),
        Tool(
            name="search_tasks",
            description="Başlığa/etikete göre görev arar.",
            inputSchema={
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    tasks = load_tasks()

    if name == "list_tasks":
        status = arguments.get("status", "open")
        due_before = arguments.get("due_before")
        out = []
        for t in tasks:
            if status == "open" and t["done"]:
                continue
            if status == "done" and not t["done"]:
                continue
            if due_before and t.get("due") and t["due"] >= due_before:
                continue
            out.append(t)
        return [TextContent(type="text", text=json.dumps(out, indent=2, ensure_ascii=False))]

    if name == "add_task":
        task = {
            "id": uuid.uuid4().hex[:8],
            "title": arguments["title"],
            "due": arguments.get("due"),
            "tags": arguments.get("tags"),
            "done": False,
            "created": datetime.utcnow().isoformat(),
        }
        tasks.append(task)
        save_tasks(tasks)
        return [TextContent(type="text", text=f"Eklendi: {task['id']} — {task['title']}")]

    if name == "complete_task":
        for t in tasks:
            if t["id"] == arguments["id"]:
                t["done"] = True
                save_tasks(tasks)
                return [TextContent(type="text", text=f"Tamamlandı: {t['title']}")]
        return [TextContent(type="text", text=f"Görev bulunamadı: {arguments['id']}")]

    if name == "search_tasks":
        q = arguments["query"].lower()
        hits = [
            t for t in tasks
            if q in t["title"].lower() or any(q in tag.lower() for tag in (t.get("tags") or []))
        ]
        return [TextContent(type="text", text=json.dumps(hits, indent=2, ensure_ascii=False))]

    raise ValueError(f"Bilinmeyen tool: {name}")


async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

python -m tasks_mcp.server

npx @modelcontextprotocol/inspector node dist/index.js
# veya Python tarafı için
npx @modelcontextprotocol/inspector python -m tasks_mcp.server

{
  "mcpServers": {
    "tasks": {
      "command": "node",
      "args": ["/full/path/to/tasks-mcp/dist/index.js"]
    }
  }
}

{
  "mcpServers": {
    "tasks": {
      "command": "python",
      "args": ["-m", "tasks_mcp.server"]
    }
  }
}