context-kit/cxk.py at main · eyalzh/context-kit · GitHub

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
import argparse
import asyncio
import logging
import sys
from importlib import metadata

from commands.create_spec import handle_create_spec
from commands.init import handle_init
from commands.mcp import (
    MCPAddHttpContext,
    MCPAddSSEContext,
    MCPAddStdioContext,
    MCPCommandContext,
    handle_mcp_command,
)
from state import State


async def async_main():
    try:
        version = metadata.version("context-kit")
    except metadata.PackageNotFoundError:
        version = "unknown"

    parser = argparse.ArgumentParser(description=f"ContextKit CLI tool (v{version})")
    parser.add_argument("--version", action="version", version=f"cxk {version}")
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    # cxk init
    subparsers.add_parser("init", help="Initialize a new ContextKit project")

    # cxk create-spec [spec-template]
    create_spec_parser = subparsers.add_parser("create-spec", help="Create spec from template")
    create_spec_parser.add_argument("spec_template", nargs=argparse.OPTIONAL, help="Path to the spec template file")
    create_spec_parser.add_argument("--output", help="Output file path (defaults to stdout if not specified)")
    create_spec_parser.add_argument("--var", action="append", help="Set template variable value (format: key=value)")
    create_spec_parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output")

    # cxk mcp
    mcp_parser = subparsers.add_parser("mcp", help="Manage MCP servers")
    mcp_subparsers = mcp_parser.add_subparsers(dest="mcp_command", help="MCP commands")

    # cxk mcp add-sse [server-name] [url]
    add_sse_parser = mcp_subparsers.add_parser("add-sse", help="Add SSE MCP server")
    add_sse_parser.add_argument("server_name", help="Name of the server")
    add_sse_parser.add_argument("url", help="URL of the SSE server")

    # cxk mcp add-stdio [server-name] --env [env-var] -- [command]
    add_stdio_parser = mcp_subparsers.add_parser("add-stdio", help="Add stdio MCP server")
    add_stdio_parser.add_argument("server_name", help="Name of the server")
    add_stdio_parser.add_argument("--env", action="append", help="Environment variable (key=value)")
    add_stdio_parser.add_argument("command_line", nargs=argparse.ONE_OR_MORE, help="Command to run")

    # cxk mcp add-http [server-name] [url] --header [header]
    add_http_parser = mcp_subparsers.add_parser("add-http", help="Add HTTP MCP server")
    add_http_parser.add_argument("server_name", help="Name of the server")
    add_http_parser.add_argument("url", help="URL of the HTTP server")
    add_http_parser.add_argument("--header", action="append", help="HTTP header (key=value)")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    if args.command == "init":
        state = State()
        await handle_init(state)

    elif args.command == "create-spec":
        state = State()
        log_level = logging.DEBUG if args.verbose else logging.WARNING
        logging.basicConfig(level=log_level, format="%(message)s", force=True)
        await handle_create_spec(args.spec_template, state, args.output, args.var)

    elif args.command == "mcp":
        state = State()
        if not args.mcp_command:
            mcp_parser.print_help()
            sys.exit(1)

        if args.mcp_command == "add-sse":
            mcp_context = MCPCommandContext(
                subcommand="add-sse",
                add_sse=MCPAddSSEContext(server_name=args.server_name, url=args.url),
            )
            await handle_mcp_command(state, mcp_context)

        elif args.mcp_command == "add-stdio":
            mcp_context = MCPCommandContext(
                subcommand="add-stdio",
                add_stdio=MCPAddStdioContext(
                    server_name=args.server_name,
                    env=args.env,
                    command=args.command_line,
                ),
            )
            await handle_mcp_command(state, mcp_context)

        elif args.mcp_command == "add-http":
            mcp_context = MCPCommandContext(
                subcommand="add-http",
                add_http=MCPAddHttpContext(server_name=args.server_name, url=args.url, headers=args.header),
            )
            await handle_mcp_command(state, mcp_context)


def main():
    """Entry point for the cxk command."""
    asyncio.run(async_main())


if __name__ == "__main__":
    main()