dalgona/examples/plugins/mcp/plugin.star at main · gosuda/dalgona · 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
load("@dal/v1", "dal")

def mcp_settings(ctx):
    raw = getattr(ctx.config, "servers_json", dal.MISSING)
    if raw == dal.MISSING or not raw:
        return {"servers": {}, "timeout_ms": 30000}
    return json.decode(raw)

def mcp_list(ctx, args):
    servers = mcp_settings(ctx).get("servers", {})
    server = args.server
    if server:
        info = servers.get(server, {})
        if not info:
            return ""
        response = ctx.mcp.call(server = server, tool = "", arguments = {})
        if getattr(response, "isError", False):
            return dal.err("mcp_error", response.text)
        if not response.text:
            return server + "  " + info.get("url", "")
        return server + "  " + info.get("url", "") + "\n" + response.text
    lines = []
    for name in sorted(servers.keys()):
        lines.append(name + "  " + servers[name].get("url", ""))
    return "\n".join(lines)

def mcp_call(ctx, args):
    arguments = json.decode(args.arguments_json)
    response = ctx.mcp.call(server = args.server, tool = args.tool, arguments = arguments)
    if getattr(response, "isError", False):
        return dal.err("mcp_error", response.text)
    return response.text

mcp_list_tool = dal.tool(
    description = "List configured MCP servers.",
    input = dal.schema(server = dal.string(default = "")),
    uses = ["mcp.call"],
    run = mcp_list,
)

mcp_call_tool = dal.tool(
    description = "Call one MCP tool with arguments_json containing a JSON object.",
    input = dal.schema(
        server = dal.string(default = ""),
        tool = dal.string(default = ""),
        arguments_json = dal.string(default = "{}"),
    ),
    uses = ["mcp.call"],
    run = mcp_call,
)

plugin = dal.plugin(
    name = "mcp",
    version = "0.1.0",
    config = dal.schema(servers_json = dal.optional(dal.string())),
    tools = {"mcp_list": mcp_list_tool, "mcp_call": mcp_call_tool},
)