feat: serve stamp tools over Streamable HTTP
/mcp on 127.0.0.1:8000 when VERAE_MCP_TRANSPORT=streamable-http.
This commit is contained in:
parent
9e3dfb6fb2
commit
e526bbf297
3 changed files with 54 additions and 6 deletions
|
|
@ -21,6 +21,12 @@ Inspector (stdio, memory ledger):
|
||||||
uv run mcp dev src/verae_mcp_timestamping_server/server.py
|
uv run mcp dev src/verae_mcp_timestamping_server/server.py
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Streamable HTTP (`http://127.0.0.1:8000/mcp`):
|
||||||
|
|
||||||
|
```sh
|
||||||
|
VERAE_MCP_TRANSPORT=streamable-http uv run verae-mcp-timestamping-server
|
||||||
|
```
|
||||||
|
|
||||||
Tests:
|
Tests:
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
|
|
|
||||||
|
|
@ -63,13 +63,13 @@ async def hash_report(
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
url = os.environ.get("VERAE_NATS_URL")
|
|
||||||
if url:
|
|
||||||
raise SystemExit(
|
|
||||||
"NATS leaf mode is wired through NatsLedger; "
|
|
||||||
"stdio Task 1 uses MemoryLedger. Set VERAE_NATS_URL on Streamable HTTP later."
|
|
||||||
)
|
|
||||||
build_mcp(MemoryLedger())
|
build_mcp(MemoryLedger())
|
||||||
|
transport = os.environ.get("VERAE_MCP_TRANSPORT", "stdio")
|
||||||
|
if transport == "streamable-http":
|
||||||
|
host = os.environ.get("VERAE_MCP_HOST", "127.0.0.1")
|
||||||
|
port = int(os.environ.get("VERAE_MCP_PORT", "8000"))
|
||||||
|
mcp.run(transport="streamable-http", host=host, port=port)
|
||||||
|
return
|
||||||
mcp.run()
|
mcp.run()
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
42
tests/test_http.py
Normal file
42
tests/test_http.py
Normal file
|
|
@ -0,0 +1,42 @@
|
||||||
|
"""Streamable HTTP: stamp over /mcp."""
|
||||||
|
|
||||||
|
import socket
|
||||||
|
|
||||||
|
import anyio
|
||||||
|
import pytest
|
||||||
|
import uvicorn
|
||||||
|
from mcp import Client
|
||||||
|
|
||||||
|
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger
|
||||||
|
from verae_mcp_timestamping_server.server import build_mcp
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.anyio
|
||||||
|
|
||||||
|
|
||||||
|
def _free_port() -> int:
|
||||||
|
sock = socket.socket()
|
||||||
|
sock.bind(("127.0.0.1", 0))
|
||||||
|
port = sock.getsockname()[1]
|
||||||
|
sock.close()
|
||||||
|
return port
|
||||||
|
|
||||||
|
|
||||||
|
async def test_stamp_over_streamable_http():
|
||||||
|
mcp = build_mcp(MemoryLedger())
|
||||||
|
app = mcp.streamable_http_app()
|
||||||
|
port = _free_port()
|
||||||
|
server = uvicorn.Server(
|
||||||
|
uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")
|
||||||
|
)
|
||||||
|
async with anyio.create_task_group() as tg:
|
||||||
|
tg.start_soon(server.serve)
|
||||||
|
with anyio.fail_after(10):
|
||||||
|
while not server.started:
|
||||||
|
await anyio.sleep(0.05)
|
||||||
|
async with Client(f"http://127.0.0.1:{port}/mcp") as client:
|
||||||
|
result = await client.call_tool(
|
||||||
|
"stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
|
||||||
|
)
|
||||||
|
assert result.is_error is not True
|
||||||
|
assert result.structured_content["receipts"][0]["hash"] == SHA256_EMPTY
|
||||||
|
server.should_exit = True
|
||||||
Loading…
Reference in a new issue