app/mcp.py

# -*- coding: utf-8 -*-
"""MCP 서버 — 에이전트가 도구로 직접 참가한다.



이게 이 챌린지의 정체성이다. Numerai도 M6도 사람이 CSV를 올린다.

참가자는 이 한 줄만 치면 된다:



    claude mcp add finchal https://<space>/mcp --header "Authorization: Bearer fc_xxx"



그러면 에이전트에게 도구 네 개가 생기고, 사람은 "챌린지 참가해줘" 한 마디만 하면 된다.



프로토콜은 JSON-RPC 2.0 위의 Streamable HTTP. 세 메서드만 구현하면 붙는다:

  initialize / tools/list / tools/call

notifications/* 는 응답 없이 202 로 받는다.

"""
import json, os, sys
from typing import Any, Dict

sys.path[:0] = [os.path.dirname(os.path.abspath(__file__))]
import i18n as I

PROTOCOL = "2025-06-18"
SERVER = {"name": "finchal", "version": "1.0.0"}

TOOLS = [
    {
        "name": "get_rules",
        "description": (
            "챌린지 규칙과 현재 시즌 정보를 읽는다. 참가 전에 반드시 먼저 호출한다. "
            "종목 목록·마감일·레버리지 상한·수수료·채점 방식이 들어 있다."),
        "inputSchema": {"type": "object", "properties": {}},
    },
    {
        "name": "get_data",
        "description": (
            "종목의 과거 시세를 받는다. 이 데이터로 직접 모델을 만들어 예측하라. "
            "미래 데이터는 주지 않는다 — 정답은 아직 세상에 없다."),
        "inputSchema": {
            "type": "object",
            "properties": {
                "asset": {"type": "string",
                          "description": "종목 코드. NVDA / BTC / GOLD / OIL"},
                "bars": {"type": "integer", "default": 500,
                         "description": "받을 봉 개수 / number of bars (max 5000)"},
                # 🔴 시간봉을 목록에 두면 안 된다. 우리 시세 피드는 일봉만 발행하는데
                #    1h 를 받아 놓고 일봉을 돌려주면 에이전트는 시간봉으로 믿고
                #    모델을 세운다. 조용히 틀린 답이 가장 나쁘다.
                "interval": {"type": "string", "enum": ["1d", "1h"], "default": "1d",
                             "description": "일봉 또는 시간봉. 채점은 시간봉 격자에서 "
                                            "이뤄집니다 / daily or hourly; scoring runs "
                                            "on the hourly grid"},
            },
            "required": ["asset"],
        },
    },
    {
        "name": "submit_position",
        "description": (
            "포지션을 제출한다. -1.0(전량 숏) ~ +1.0(전량 롱). 레버리지는 1로 고정이며 "
            "범위를 벗어나면 잘린다. 다시 제출할 때까지 이 포지션이 유지된다. "
            "포지션을 바꿀 때마다 수수료가 나가므로 자주 뒤집으면 손해다."),
        "inputSchema": {
            "type": "object",
            "properties": {
                "asset": {"type": "string"},
                "position": {"type": "number", "minimum": -1, "maximum": 1},
                "note": {"type": "string", "description": "선택. 판단 근거 메모"},
            },
            "required": ["asset", "position"],
        },
    },
    {
        "name": "check_score",
        "description": (
            "내 성적과 순위를 확인한다. 수익률·순위·운의 한계선 대비 위치를 돌려준다."),
        "inputSchema": {
            "type": "object",
            "properties": {"asset": {"type": "string",
                                     "description": "생략하면 참가 중인 전 종목"}},
        },
    },
]


def _ok(rid, result):
    return {"jsonrpc": "2.0", "id": rid, "result": result}


def _err(rid, code, msg):
    return {"jsonrpc": "2.0", "id": rid, "error": {"code": code, "message": msg}}


def _text(obj):
    """도구 결과는 텍스트 콘텐츠로 돌려준다. 에이전트가 JSON을 잘 읽으므로

    구조를 그대로 직렬화하되, 사람이 로그를 볼 때도 읽히게 들여쓴다."""
    return {"content": [{"type": "text",
                         "text": json.dumps(obj, ensure_ascii=False, indent=1)}]}


def tools(lang="ko"):
    """도구 목록. 설명이 언어를 탄다 — 에이전트는 이 설명을 읽고 판단한다."""
    out = []
    for t in TOOLS:
        d = dict(t)
        d["description"] = I.t(I.TOOLDESC.get(t["name"], t["description"]), lang)
        out.append(d)
    return out


def handle(body: Dict[str, Any], call, lang: str = "ko") -> Any:
    """call(name, args) -> dict 를 받아 도구를 실제로 실행한다.



    반환이 None 이면 응답 없음(notification)이라는 뜻이다.

    """
    method = body.get("method")
    rid = body.get("id")

    if method == "initialize":
        return _ok(rid, {
            "protocolVersion": PROTOCOL,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": SERVER,
            "instructions": I.t(I.MCP_INSTRUCTIONS, lang),
        })

    if method and method.startswith("notifications/"):
        return None                      # 알림은 응답하지 않는다

    if method == "tools/list":
        return _ok(rid, {"tools": tools(lang)})

    if method == "tools/call":
        p = body.get("params") or {}
        name = p.get("name")
        args = p.get("arguments") or {}
        if not any(t["name"] == name for t in TOOLS):
            return _err(rid, -32602, "알 수 없는 도구: %s" % name)
        try:
            return _ok(rid, _text(call(name, args)))
        except PermissionError as e:
            return _ok(rid, {"content": [{"type": "text", "text": str(e)}],
                             "isError": True})
        except Exception as e:
            return _ok(rid, {"content": [
                {"type": "text", "text": "%s: %s" % (type(e).__name__, e)}],
                "isError": True})

    if method == "ping":
        return _ok(rid, {})

    return _err(rid, -32601, "지원하지 않는 메서드: %s" % method)