vtcode-core 0.104.1

Core library for VT Code - a Rust-based terminal coding agent
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

#!/usr/bin/env python3
"""Minimal newline-delimited JSON-RPC MCP mock server for integration tests."""

from __future__ import annotations

import json
import sys
from typing import Any


def send(message: dict[str, Any]) -> None:
    sys.stdout.write(json.dumps(message))
    sys.stdout.write("\n")
    sys.stdout.flush()


def initialize_result(protocol_version: str) -> dict[str, Any]:
    return {
        "protocolVersion": protocol_version,
        "capabilities": {
            "tools": {
                "listChanged": False,
            }
        },
        "serverInfo": {
            "name": "vtcode-mock-mcp",
            "version": "0.1.0",
        },
    }


def list_tools_result() -> dict[str, Any]:
    return {
        "tools": [
            {
                "name": "echo",
                "description": "Echoes back the provided message",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "message": {
                            "type": "string",
                        }
                    },
                    "required": ["message"],
                },
            }
        ]
    }


def call_tool_result(method_params: dict[str, Any]) -> dict[str, Any]:
    tool_name = method_params.get("name")
    arguments = method_params.get("arguments") or {}

    if tool_name != "echo":
        return {
            "isError": True,
            "content": [
                {
                    "type": "text",
                    "text": f"unknown tool: {tool_name}",
                }
            ],
        }

    message = str(arguments.get("message", ""))
    return {
        "content": [
            {
                "type": "text",
                "text": f"echo:{message}",
            }
        ]
    }


for raw_line in sys.stdin:
    line = raw_line.strip()
    if not line:
        continue

    try:
        request = json.loads(line)
    except json.JSONDecodeError:
        continue

    method = request.get("method")
    request_id = request.get("id")
    params = request.get("params") or {}

    # Notifications have no id and require no response.
    if request_id is None:
        continue

    if method == "initialize":
        protocol = str(params.get("protocolVersion", "2024-11-05"))
        send(
            {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": initialize_result(protocol),
            }
        )
    elif method == "tools/list":
        send(
            {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": list_tools_result(),
            }
        )
    elif method == "tools/call":
        send(
            {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": call_tool_result(params),
            }
        )
    elif method == "ping":
        send({"jsonrpc": "2.0", "id": request_id, "result": {}})
    else:
        send(
            {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {
                    "code": -32601,
                    "message": f"Method not found: {method}",
                },
            }
        )