#!/usr/bin/env python3 """「我的今天」任务 MCP server（stdio）。 用个人 API Token 代替本人读写自己的任务，能力面就是 my-day 页面上你自己能做的 那些事：看当天看板、记任务、推进状态、改字段。刻意不提供任何团队/管理类工具—— token 在服务端也拿不到那些接口，这里不做第二套权限，只是不给多余的入口。 和 workforce-local-mcp.py 的区别：那个是只读本机脱敏快照（Shadow Read Only）， 这个是代表用户本人写自己的任务，两者用的凭据、能碰的数据完全不同。 用法： workforce-tasks-mcp.py --base-url https://cc.jiejingyun.org --token-file ~/.workforce/api-token """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.request from pathlib import Path from typing import Any TIMEOUT = 20 def api(base_url: str, token: str, path: str, method: str = "GET", body: dict[str, Any] | None = None) -> Any: data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request( f"{base_url.rstrip('/')}/api/v1{path}", data=data, method=method, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace") try: detail = json.loads(detail).get("detail", detail) except Exception: pass raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"无法连接服务端: {exc.reason}") from exc def tool_result(payload: Any, *, is_error: bool = False) -> dict[str, Any]: return { "content": [{"type": "text", "text": json.dumps(payload, ensure_ascii=False, indent=2)}], "isError": is_error, } def _slim(task: dict[str, Any]) -> dict[str, Any]: """只回模型用得上的字段，别把整行数据库记录塞进上下文。 服务端有两种任务形态：/my-work 经过映射（title / 大写 status）， 创建和状态流转接口返回原始行（content / 小写 status）。这里统一成一种， 否则同一个工具集里「同一个任务」会长出两副样子，模型必然搞混。 """ status = str(task.get("status") or "").upper() return { "id": task.get("id"), "title": task.get("title") or task.get("content"), "status": status or None, "priority": task.get("priority"), "due_date": task.get("due_date"), "task_date": task.get("task_date"), "type": task.get("type"), "note": task.get("description") or task.get("note") or "", "is_focus": bool(task.get("is_today_focus")), } TOOLS = [ { "name": "get_my_day", "description": "看某一天的任务看板：待开始/进行中/已完成，外加未排期任务。不传日期就是今天。", "inputSchema": { "type": "object", "properties": {"date": {"type": "string", "description": "YYYY-MM-DD，留空为今天"}}, "additionalProperties": False, }, }, { "name": "add_tasks", "description": "记一条或多条任务。scheduled=false 表示先不排期（进未排期区），不写死在某一天。", "inputSchema": { "type": "object", "properties": { "items": { "type": "array", "minItems": 1, "maxItems": 20, "items": { "type": "object", "properties": { "content": {"type": "string", "description": "任务标题，必填"}, "priority": {"type": "string", "enum": ["high", "normal", "low"]}, "due_date": {"type": "string", "description": "YYYY-MM-DD"}, "note": {"type": "string"}, "scheduled": {"type": "boolean", "description": "true=排到今天（默认），false=不排期"}, "task_date": {"type": "string", "description": "YYYY-MM-DD，排到指定某天"}, }, "required": ["content"], "additionalProperties": False, }, } }, "required": ["items"], "additionalProperties": False, }, }, { "name": "set_task_status", "description": "推进任务状态：start=开始，complete=完成，reopen=移回待开始，cancel=取消。", "inputSchema": { "type": "object", "properties": { "task_id": {"type": "integer"}, "action": {"type": "string", "enum": ["start", "complete", "reopen", "cancel"]}, }, "required": ["task_id", "action"], "additionalProperties": False, }, }, { "name": "update_task", "description": "改任务的优先级/截止日期/备注/标题/排期。团队派发的任务标题改不了，服务端会拒。", "inputSchema": { "type": "object", "properties": { "task_id": {"type": "integer"}, "content": {"type": "string"}, "priority": {"type": "string", "enum": ["high", "normal", "low"]}, "due_date": {"type": "string", "description": "YYYY-MM-DD"}, "clear_due_date": {"type": "boolean"}, "note": {"type": "string"}, "scheduled": {"type": "boolean", "description": "true=排到今天，false=退回未排期"}, }, "required": ["task_id"], "additionalProperties": False, }, }, { "name": "set_today_focus", "description": "把某条任务设为今日重点；clear=true 表示取消重点。每天只有一个重点。", "inputSchema": { "type": "object", "properties": {"task_id": {"type": "integer"}, "clear": {"type": "boolean"}}, "required": ["task_id"], "additionalProperties": False, }, }, ] def call_tool(name: str, args: dict[str, Any], base_url: str, token: str) -> dict[str, Any]: if name == "get_my_day": day = (args.get("date") or "").strip() d = api(base_url, token, f"/my-work{f'?date_str={day}' if day else ''}") tasks = d.get("tasks") or {} focus = d.get("focus") return tool_result({ "date": d.get("date"), "focus": _slim(focus) if focus else None, "todo": [_slim(t) for t in tasks.get("todo", [])], "in_progress": [_slim(t) for t in tasks.get("in_progress", [])], "done": [_slim(t) for t in tasks.get("done", [])], "unscheduled": [_slim(t) for t in (d.get("backlog") or [])], "summary": d.get("summary"), }) if name == "add_tasks": items = args.get("items") or [] created, failed = [], [] for item in items: content = str(item.get("content") or "").strip() if not content: failed.append({"content": item.get("content"), "error": "标题不能为空"}) continue body: dict[str, Any] = {"content": content, "scheduled": item.get("scheduled", True)} for k in ("priority", "due_date", "note", "task_date"): if item.get(k): body[k] = item[k] try: created.append(_slim(api(base_url, token, "/employee/tasks", "POST", body))) except RuntimeError as exc: # 逐条建，中途失败不影响已经建好的；把没建成的原样报回去 failed.append({"content": content, "error": str(exc)}) return tool_result({"created": created, "failed": failed}, is_error=bool(failed and not created)) if name == "set_task_status": task_id = int(args["task_id"]) action = str(args["action"]) if action not in {"start", "complete", "reopen", "cancel"}: return tool_result({"error": f"不支持的动作: {action}"}, is_error=True) return tool_result(_slim(api(base_url, token, f"/work-items/{task_id}/{action}", "POST", {}))) if name == "update_task": task_id = int(args["task_id"]) body = {k: args[k] for k in ("content", "priority", "due_date", "clear_due_date", "note", "scheduled") if k in args and args[k] is not None} if not body: return tool_result({"error": "没有要改的字段"}, is_error=True) return tool_result(_slim(api(base_url, token, f"/employee/tasks/{task_id}", "PATCH", body))) if name == "set_today_focus": task_id = int(args["task_id"]) clear = "?clear=true" if args.get("clear") else "" return tool_result(api(base_url, token, f"/work-items/{task_id}/focus{clear}", "POST", {})) return tool_result({"error": f"未知工具: {name}"}, is_error=True) def response(mid: Any, result: dict[str, Any]) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": mid, "result": result} def error(mid: Any, code: int, message: str) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": mid, "error": {"code": code, "message": message}} def handle(message: dict[str, Any], base_url: str, token: str) -> dict[str, Any] | None: method = message.get("method") mid = message.get("id") params = message.get("params") if isinstance(message.get("params"), dict) else {} if method == "notifications/initialized": return None if method == "initialize": return response(mid, { "protocolVersion": str(params.get("protocolVersion") or "2025-03-26"), "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "workforce-tasks-mcp", "version": "1.0.0"}, "instructions": ("代表用户本人读写他自己的任务。只能碰本人任务，" "没有任何团队管理、设备或后台能力。"), }) if method == "ping": return response(mid, {}) if method == "tools/list": return response(mid, {"tools": TOOLS}) if method == "tools/call": try: name = str(params.get("name") or "") args = params.get("arguments") if isinstance(params.get("arguments"), dict) else {} return response(mid, call_tool(name, args, base_url, token)) except Exception as exc: return response(mid, tool_result({"error": str(exc)}, is_error=True)) return error(mid, -32601, "method not found") def main() -> int: parser = argparse.ArgumentParser(description="我的今天 · 任务 MCP server") parser.add_argument("--base-url", required=True, help="例如 https://cc.jiejingyun.org") parser.add_argument("--token-file", required=True, help="存放个人 API Token 的文件") args = parser.parse_args() try: token = Path(args.token_file).expanduser().read_text(encoding="utf-8").strip() except OSError: print("读不到 API Token 文件", file=sys.stderr) return 2 if not token.startswith("wtk_"): print("API Token 格式不对（应以 wtk_ 开头）", file=sys.stderr) return 2 for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) except json.JSONDecodeError: continue if not isinstance(message, dict) or message.get("jsonrpc") != "2.0": continue out = handle(message, args.base_url, token) if out is not None: sys.stdout.write(json.dumps(out, ensure_ascii=False) + "\n") sys.stdout.flush() return 0 if __name__ == "__main__": raise SystemExit(main())