Tom FluxandClaude Opus 4.8
44de784f70
A standalone MCP server (no Django import) exposing 7 tools over Streamable HTTP, backed by the Django REST API via the caine token. - mcp_server/: client.py (httpx wrapper over /api/), server.py (the tools + FastMCP app + main), __main__.py, tests.py. - Tools: get_pantry, set_item_state, add_to_pantry, what_can_i_cook, get_recipes, log_cook (suggests, never mutates), create_meta_recipe (brainstorm -> commit). Each description says when to call it. - deploy/food-mcp.service: systemd unit (own process, runs .venv python -m mcp_server, loads /var/lib/food/.env). - deploy/food.tomflux.xyz.nginx: current config + an authless /mcp/<secret>/ location proxying to 127.0.0.1:8765 (the URL secret is the credential; SSE-friendly buffering/timeout). - pyproject: [dependency-groups] mcp = [fastmcp, httpx]; deploy with `uv sync --group mcp`. Verified: 7 tools register on fastmcp 3.x, run() accepts transport/ host/port/path, 6 FoodClient unit tests pass (httpx MockTransport). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""Tests for the FoodClient HTTP layer, using httpx.MockTransport (no network,
|
|
no Django, no fastmcp). Run with: python -m unittest mcp_server.tests
|
|
"""
|
|
import json
|
|
import unittest
|
|
|
|
import httpx
|
|
|
|
from mcp_server.client import FoodClient, FoodApiError
|
|
|
|
|
|
def make_client(handler):
|
|
transport = httpx.MockTransport(handler)
|
|
http = httpx.Client(
|
|
transport=transport,
|
|
base_url="http://api.test/api",
|
|
headers={"Authorization": "Token x"},
|
|
)
|
|
return FoodClient(http=http)
|
|
|
|
|
|
class FoodClientTests(unittest.TestCase):
|
|
def test_set_state_posts_expected_body(self):
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["path"] = request.url.path
|
|
captured["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"state": "out"})
|
|
|
|
out = make_client(handler).set_state("noodles", "out", location="cupboard")
|
|
self.assertEqual(captured["path"], "/api/pantry/set-state/")
|
|
self.assertEqual(
|
|
captured["body"],
|
|
{"ingredient": "noodles", "state": "out", "location": "cupboard"},
|
|
)
|
|
self.assertEqual(out["state"], "out")
|
|
|
|
def test_set_state_omits_location_when_absent(self):
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={})
|
|
|
|
make_client(handler).set_state("eggs", "in")
|
|
self.assertNotIn("location", captured["body"])
|
|
|
|
def test_bulk_add_wraps_items(self):
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["body"] = json.loads(request.content)
|
|
return httpx.Response(201, json={"added": 1})
|
|
|
|
make_client(handler).bulk_add([{"ingredient_name": "eggs"}])
|
|
self.assertEqual(captured["body"], {"items": [{"ingredient_name": "eggs"}]})
|
|
|
|
def test_what_can_i_cook_passes_servings(self):
|
|
captured = {}
|
|
|
|
def handler(request):
|
|
captured["servings"] = request.url.params.get("servings")
|
|
return httpx.Response(200, json={"results": []})
|
|
|
|
make_client(handler).what_can_i_cook(servings=4)
|
|
self.assertEqual(captured["servings"], "4")
|
|
|
|
def test_error_response_raises(self):
|
|
def handler(request):
|
|
return httpx.Response(404, json={"error": "nope"})
|
|
|
|
with self.assertRaises(FoodApiError):
|
|
make_client(handler).set_state("x", "out")
|
|
|
|
def test_unreachable_api_raises(self):
|
|
def handler(request):
|
|
raise httpx.ConnectError("boom")
|
|
|
|
with self.assertRaises(FoodApiError):
|
|
make_client(handler).pantry()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|