kyleandClaude Sonnet 5.5
167f53ceeb
The MCP server is now a ROS node and the only way in or out, per the owner's decision that ROS never leaves the robot. It serves MCP on the USB address only with a bearer token, keeps the stable tool name look() (which waits for the next camera frame, so it is never stale, and passes the camera's JPEG through), and registers turn_head/head_position only when a head is fitted, which today it is not. While an authenticated client is connected it republishes the agent heartbeat on /hermes/state, so the face wakes on connect and sleeps about 40 s after the client leaves; look() also makes the face squint. The token is generated on the Pi into /etc/robot-mcp.env (root:hermes, 640) and never printed. Verified end to end from talos with a real MCP client over the USB link: tools listed, a 157 KB JPEG back in 0.08 s, a wrong token refused, the Wi-Fi path dropped by the firewall. Host config snippets now point at 10.55.0.2. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
import asyncio
|
|
import base64
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip('mcp')
|
|
from starlette.testclient import TestClient # noqa: E402
|
|
|
|
from hermes_mcp.bridge import Activity, BearerAuth, LatestFrame, build_app, build_mcp # noqa: E402
|
|
|
|
TOKEN = 'x' * 32
|
|
|
|
|
|
class FakeHead:
|
|
def move(self, pan, tilt):
|
|
return max(-80, min(80, pan)), max(-45, min(45, tilt))
|
|
|
|
def position(self):
|
|
return 0.0, 0.0
|
|
|
|
|
|
def tool_names(mcp):
|
|
return {t.name for t in asyncio.run(mcp.list_tools())}
|
|
|
|
|
|
def test_tools_are_look_only_without_a_head_and_keep_their_names_with_one():
|
|
assert tool_names(build_mcp(LatestFrame())) == {'look'}
|
|
assert tool_names(build_mcp(LatestFrame(), head=FakeHead())) == {'look', 'turn_head', 'head_position'}
|
|
|
|
|
|
def test_look_returns_the_next_frame_not_a_stale_one():
|
|
frames = LatestFrame()
|
|
frames.put(b'stale', time.monotonic() - 5)
|
|
announced = []
|
|
mcp = build_mcp(frames, announce_look=lambda: announced.append(1), look_timeout_s=2.0)
|
|
threading.Timer(0.3, lambda: frames.put(b'\xff\xd8fresh')).start()
|
|
result = asyncio.run(mcp.call_tool('look', {}))
|
|
image = result.content[0]
|
|
assert announced == [1]
|
|
assert image.type == 'image' and image.mime_type == 'image/jpeg'
|
|
assert base64.b64decode(image.data) == b'\xff\xd8fresh' # the frame that arrived after the call
|
|
|
|
|
|
def test_look_raises_when_the_camera_is_silent():
|
|
mcp = build_mcp(LatestFrame(), look_timeout_s=0.3)
|
|
with pytest.raises(Exception) as err:
|
|
asyncio.run(mcp.call_tool('look', {}))
|
|
assert 'camera' in str(err.value.__cause__) # the SDK wraps it; the cause is ours
|
|
|
|
|
|
def test_wait_newer_semantics():
|
|
f = LatestFrame()
|
|
t0 = time.monotonic()
|
|
assert f.wait_newer(t0, 0.1) is None # nothing yet
|
|
f.put(b'a', t0 - 1)
|
|
assert f.wait_newer(t0, 0.1) is None # older than t0
|
|
threading.Timer(0.1, lambda: f.put(b'b')).start()
|
|
assert f.wait_newer(t0, 1.0) == b'b'
|
|
|
|
|
|
def test_auth_rejects_missing_and_wrong_tokens_and_counts_only_good_requests():
|
|
app, activity = build_app(build_mcp(LatestFrame()), TOKEN, '127.0.0.1')
|
|
with TestClient(app) as client:
|
|
assert client.get('/mcp').status_code == 401
|
|
assert client.get('/mcp', headers={'Authorization': 'Bearer wrong'}).status_code == 401
|
|
assert not activity.connected(time.monotonic(), 30.0) # rejected requests are not a client
|
|
r = client.get('/mcp', headers={'Authorization': f'Bearer {TOKEN}'})
|
|
assert r.status_code != 401
|
|
assert activity.connected(time.monotonic(), 30.0) # an authenticated one is
|
|
assert not activity.connected(time.monotonic() + 31.0, 30.0) # ...until the grace runs out
|
|
|
|
|
|
def test_activity_counts_inflight_requests_as_connected():
|
|
async def run():
|
|
release = asyncio.Event()
|
|
|
|
async def slow_app(scope, receive, send):
|
|
await release.wait()
|
|
|
|
act = Activity(slow_app)
|
|
task = asyncio.create_task(act({'type': 'http'}, None, None))
|
|
await asyncio.sleep(0.05)
|
|
assert act.inflight == 1 and act.connected(time.monotonic() + 1000, 1.0) # a held stream stays connected
|
|
release.set()
|
|
await task
|
|
assert act.inflight == 0
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_bearer_auth_passes_lifespan_through():
|
|
seen = []
|
|
|
|
async def inner(scope, receive, send):
|
|
seen.append(scope['type'])
|
|
|
|
asyncio.run(BearerAuth(inner, TOKEN)({'type': 'lifespan'}, None, None))
|
|
assert seen == ['lifespan']
|