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
|
-- Self-check for panto-web. Run with: panto lua _selfcheck.lua
local registered = {}
local json = require("panto").ext.json
package.loaded["panto"] = {
ext = {
register_tool = function(tool) registered[#registered + 1] = tool end,
json = json,
},
}
local here = debug.getinfo(1, "S").source:match("^@(.*)/[^/]+$") or "."
package.path = here .. "/panto-web/?.lua;" .. here .. "/panto-web/?/init.lua;" .. package.path
local web = require("init")
assert(web.name == "web", "entry name")
web.activate()
assert(registered[1].name == "web.fetch", "fetch tool registered")
assert(registered[2].name == "web.search", "search tool registered")
assert(web.on_fetch({}):find("non-empty", 1, true), "fetch rejects missing URL")
assert(web.on_fetch({ url = "" }):find("non-empty", 1, true), "fetch rejects empty URL")
assert(web.on_search({ query = "" }):find("non-empty", 1, true), "search rejects empty query")
assert(web.on_search({ query = "ok", num_results = 0 }):find("integer", 1, true), "search rejects count")
local markdown = web.build_request(7, "tools/call",
web.tool_params("markdown", { url = "https://example.com" }))
assert(markdown.id == 7 and markdown.params.name == "markdown", "markdown MCP request")
local evaluate = web.build_request(8, "tools/call",
web.tool_params("evaluate", { script = "1 + 1" }))
assert(evaluate.id == 8 and evaluate.params.name == "evaluate", "evaluate MCP request")
-- Replies must resume only the coroutine waiting for their id.
local resumed = {}
local fake = { pending = {}, buf = "" }
for _, id in ipairs({ 1, 2 }) do
local co = coroutine.create(function()
coroutine.yield()
resumed[#resumed + 1] = id
end)
assert(coroutine.resume(co))
fake.pending[id] = { co = co }
end
assert(not web.feed(fake, json.encode({ jsonrpc = "2.0", id = 2, result = {} }) .. "\n"))
assert(#resumed == 1 and resumed[1] == 2 and fake.pending[1], "response routed by id")
assert(not web.feed(fake, json.encode({ jsonrpc = "2.0", id = 1, result = {} }) .. "\n"))
assert(#resumed == 2 and resumed[2] == 1, "second response routed by id")
local body = json.decode(web.exa_body("Lightpanda", 3))
assert(body.query == "Lightpanda" and body.numResults == 3 and body.type == "auto", "Exa request fields")
assert(body.contents.highlights == true, "Exa highlights requested")
local mcp_body = json.decode(web.exa_mcp_body("Lightpanda", 3))
assert(mcp_body.params.name == "web_search_exa", "keyless Exa MCP tool")
assert(mcp_body.params.arguments.query == "Lightpanda" and mcp_body.params.arguments.numResults == 3,
"keyless Exa MCP arguments")
local mcp_payload = json.encode({ jsonrpc = "2.0", id = 1, result = {
content = { { type = "text", text = "Title: Result" } }, isError = false,
} })
assert(web.exa_mcp_result("event: message\ndata: " .. mcp_payload .. "\n") == "Title: Result",
"keyless Exa SSE result")
local fake_key = "fake-secret-key"
assert(not web.exa_body("Lightpanda", 3):find(fake_key, 1, true), "key absent from Exa body")
assert(not web.scrub("failed with " .. fake_key, fake_key):find(fake_key, 1, true), "key redacted from output")
local oversized = web.truncate(string.rep("x", 101 * 1024))
assert(#oversized < 101 * 1024 and oversized:find("truncated", 1, true), "output capped")
local text = assert(web.result_text({ content = {
{ type = "text", text = "first" }, { type = "text", text = "second" },
} }))
assert(text == "first\nsecond", "MCP text extracted")
local _, tool_err = web.result_text({ isError = true, content = { { type = "text", text = "failed" } } })
assert(tool_err == "failed", "MCP tool error extracted")
print("panto-web selfcheck: OK")
|
