diff --git a/docs/LLM Tool Server.md b/docs/LLM Tool Server.md new file mode 100644 index 00000000..68f16499 --- /dev/null +++ b/docs/LLM Tool Server.md @@ -0,0 +1,84 @@ +# LLM Tool Server + +The `obd_ai` package includes a lightweight, runnable server entrypoint for exposing the safe read-only diagnostic tools to external agents. + +This transport is function-calling-friendly JSON over HTTP and reuses the same constrained tool surface used by the in-process Python API. + +## Exposed tools + +- `connect_vehicle` +- `get_vehicle_status` +- `list_supported_commands` +- `read_sensor` +- `read_sensors` +- `get_dtc_codes` +- `get_freeze_frame` +- `get_emissions_monitor_status` +- `basic_health_scan` + +These tools remain read-only. The server does **not** expose raw OBD command construction or write operations. + +## Start the server + +```bash +python3 -m obd_ai.server serve-http --host 127.0.0.1 --port 8000 +``` + +Or, after installing the project in editable or packaged form: + +```bash +obd-ai-server serve-http --host 127.0.0.1 --port 8000 +``` + +## Inspect the tool catalog + +```bash +python3 -m obd_ai.server list-tools +``` + +## Make a one-shot tool call from the CLI + +```bash +python3 -m obd_ai.server call connect_vehicle --input-json '{"portstr":"/dev/ttyUSB0"}' +python3 -m obd_ai.server call basic_health_scan +``` + +## HTTP endpoints + +### `GET /health` + +Returns a small readiness payload: + +```json +{ + "ok": true, + "service": "obd_ai_server" +} +``` + +### `GET /tools` + +Returns the exposed tool definitions and input schemas. + +### `POST /call` + +Request body: + +```json +{ + "tool": "read_sensor", + "input": { + "command_key": "engine_rpm" + } +} +``` + +### `POST /tools/` + +Equivalent convenience route for directly calling a specific tool. + +## Notes + +- Session lifecycle is managed by the underlying tool surface. +- Call `connect_vehicle` before issuing read operations. +- Error payloads are returned in structured JSON and preserve existing `obd_ai` error codes when possible. diff --git a/mkdocs.yml b/mkdocs.yml index ad9086bb..927dc1ca 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -9,6 +9,7 @@ nav: - 'Command Lookup': 'Command Lookup.md' - 'Command Tables' : 'Command Tables.md' - 'Responses': 'Responses.md' +- 'LLM Tool Server': 'LLM Tool Server.md' - 'Async Connections': 'Async Connections.md' - 'Custom Commands': 'Custom Commands.md' - 'Debug': 'Debug.md' diff --git a/obd_ai/server.py b/obd_ai/server.py new file mode 100644 index 00000000..490b7b57 --- /dev/null +++ b/obd_ai/server.py @@ -0,0 +1,338 @@ +"""Runnable function-calling-compatible server entrypoint for ``obd_ai``.""" + +from __future__ import annotations + +import argparse +import json +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Dict, List, Mapping, Optional + +from .tools import OBDAIReadOnlyToolSurface + + +_TOOL_DEFINITIONS: List[Dict[str, Any]] = [ + { + "name": "connect_vehicle", + "description": "Open a read-only OBD session and return connection metadata.", + "input_schema": { + "type": "object", + "properties": { + "portstr": {"type": "string"}, + "baudrate": {"type": "integer"}, + "protocol": {"type": "string"}, + "fast": {"type": "boolean"}, + "timeout": {"type": "number"}, + "check_voltage": {"type": "boolean"}, + "start_low_power": {"type": "boolean"}, + }, + "additionalProperties": False, + }, + }, + { + "name": "get_vehicle_status", + "description": "Return current connection and protocol status for the active session.", + "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "list_supported_commands", + "description": "List approved diagnostic commands, optionally filtered to supported ones.", + "input_schema": { + "type": "object", + "properties": { + "include_support": {"type": "boolean"}, + "only_supported": {"type": "boolean"}, + }, + "additionalProperties": False, + }, + }, + { + "name": "read_sensor", + "description": "Read one approved sensor by catalog key.", + "input_schema": { + "type": "object", + "properties": { + "command_key": {"type": "string"}, + }, + "required": ["command_key"], + "additionalProperties": False, + }, + }, + { + "name": "read_sensors", + "description": "Read multiple approved sensors in one call.", + "input_schema": { + "type": "object", + "properties": { + "command_keys": { + "type": "array", + "items": {"type": "string"}, + } + }, + "required": ["command_keys"], + "additionalProperties": False, + }, + }, + { + "name": "get_dtc_codes", + "description": "Read stored and pending diagnostic trouble codes.", + "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "get_freeze_frame", + "description": "Read the freeze-frame trigger DTC when available.", + "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "get_emissions_monitor_status", + "description": "Read monitor readiness and MIL status.", + "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "basic_health_scan", + "description": "Run a deterministic, read-only baseline vehicle health scan.", + "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, +] + + +def list_tool_definitions() -> List[Dict[str, Any]]: + """Return the server's exposed tool definitions.""" + + return [dict(tool) for tool in _TOOL_DEFINITIONS] + + +class OBDAIFunctionServer: + """Thin tool registry around ``OBDAIReadOnlyToolSurface``.""" + + def __init__(self, surface: Optional[OBDAIReadOnlyToolSurface] = None): + self._surface = surface or OBDAIReadOnlyToolSurface() + self._tools = { + tool["name"]: getattr(self._surface, tool["name"]) + for tool in _TOOL_DEFINITIONS + } + + def list_tools(self) -> Dict[str, Any]: + return { + "ok": True, + "tools": list_tool_definitions(), + "count": len(_TOOL_DEFINITIONS), + } + + def call_tool(self, name: str, input_payload: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + tool = self._tools.get(name) + if tool is None: + return { + "ok": False, + "error": { + "code": "unknown_tool", + "message": f"Unknown tool: {name}", + "details": {"tool": name}, + }, + } + + try: + return tool(input_payload) + except Exception as exc: # pragma: no cover + return { + "ok": False, + "error": { + "code": "server_error", + "message": "Unhandled tool server exception.", + "details": { + "tool": name, + "exception_type": exc.__class__.__name__, + "exception": str(exc), + }, + }, + } + + def close(self) -> None: + self._surface.close() + + +class _OBDAIHTTPRequestHandler(BaseHTTPRequestHandler): + server_version = "OBDAIServer/0.1" + + def do_GET(self) -> None: # noqa: N802 + if self.path == "/health": + self._send_json(HTTPStatus.OK, {"ok": True, "service": "obd_ai_server"}) + return + + if self.path == "/tools": + self._send_json(HTTPStatus.OK, self.server.app.list_tools()) + return + + self._send_json(HTTPStatus.NOT_FOUND, _not_found_payload(self.path)) + + def do_POST(self) -> None: # noqa: N802 + payload = self._read_json_body() + if isinstance(payload, tuple): + status, body = payload + self._send_json(status, body) + return + + if self.path == "/call": + tool_name = payload.get("tool") + input_payload = payload.get("input", {}) + + if not isinstance(tool_name, str) or not tool_name: + self._send_json( + HTTPStatus.BAD_REQUEST, + { + "ok": False, + "error": { + "code": "invalid_request", + "message": "POST /call requires a non-empty string tool field.", + "details": {}, + }, + }, + ) + return + + status = HTTPStatus.OK + result = self.server.app.call_tool(tool_name, input_payload) + if not result.get("ok") and result.get("error", {}).get("code") == "unknown_tool": + status = HTTPStatus.NOT_FOUND + self._send_json(status, result) + return + + if self.path.startswith("/tools/"): + tool_name = self.path.rsplit("/", 1)[-1] + result = self.server.app.call_tool(tool_name, payload) + status = HTTPStatus.OK + if not result.get("ok") and result.get("error", {}).get("code") == "unknown_tool": + status = HTTPStatus.NOT_FOUND + self._send_json(status, result) + return + + self._send_json(HTTPStatus.NOT_FOUND, _not_found_payload(self.path)) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A003 + return + + def _read_json_body(self): + raw_length = self.headers.get("Content-Length", "0") + try: + length = int(raw_length) + except ValueError: + length = 0 + + raw_body = self.rfile.read(length) if length > 0 else b"{}" + try: + payload = json.loads(raw_body.decode("utf-8")) + except json.JSONDecodeError as exc: + return ( + HTTPStatus.BAD_REQUEST, + { + "ok": False, + "error": { + "code": "invalid_json", + "message": "Request body must be valid JSON.", + "details": {"error": str(exc)}, + }, + }, + ) + + if not isinstance(payload, dict): + return ( + HTTPStatus.BAD_REQUEST, + { + "ok": False, + "error": { + "code": "invalid_request", + "message": "Request body must be a JSON object.", + "details": {}, + }, + }, + ) + + return payload + + def _send_json(self, status: HTTPStatus, payload: Mapping[str, Any]) -> None: + body = json.dumps(payload, sort_keys=True).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +def serve_http( + host: str = "127.0.0.1", + port: int = 8000, + surface: Optional[OBDAIReadOnlyToolSurface] = None, +) -> ThreadingHTTPServer: + """Create a threaded HTTP server exposing the read-only tool surface.""" + + app = OBDAIFunctionServer(surface=surface) + server = ThreadingHTTPServer((host, port), _OBDAIHTTPRequestHandler) + server.app = app # type: ignore[attr-defined] + return server + + +def main(argv: Optional[List[str]] = None) -> int: + parser = argparse.ArgumentParser(description="Run the obd_ai tool server.") + subparsers = parser.add_subparsers(dest="command", required=True) + + subparsers.add_parser("list-tools", help="Print exposed tool definitions as JSON.") + + call_parser = subparsers.add_parser("call", help="Call one tool once and print JSON.") + call_parser.add_argument("tool", help="Tool name to invoke.") + call_parser.add_argument( + "--input-json", + default="{}", + help="JSON object to pass as the tool input.", + ) + + serve_parser = subparsers.add_parser("serve-http", help="Run the HTTP tool server.") + serve_parser.add_argument("--host", default="127.0.0.1") + serve_parser.add_argument("--port", type=int, default=8000) + + args = parser.parse_args(argv) + + if args.command == "list-tools": + print(json.dumps({"ok": True, "tools": list_tool_definitions()}, indent=2, sort_keys=True)) + return 0 + + if args.command == "call": + payload = json.loads(args.input_json) + if not isinstance(payload, dict): + raise SystemExit("--input-json must decode to a JSON object") + app = OBDAIFunctionServer() + try: + print(json.dumps(app.call_tool(args.tool, payload), indent=2, sort_keys=True)) + finally: + app.close() + return 0 + + server = serve_http(host=args.host, port=args.port) + try: + print(json.dumps({ + "ok": True, + "service": "obd_ai_server", + "transport": "http", + "host": args.host, + "port": args.port, + }, sort_keys=True)) + server.serve_forever() + except KeyboardInterrupt: # pragma: no cover + return 0 + finally: + server.server_close() + server.app.close() # type: ignore[attr-defined] + + +def _not_found_payload(path: str) -> Dict[str, Any]: + return { + "ok": False, + "error": { + "code": "not_found", + "message": f"No route for path: {path}", + "details": {"path": path}, + }, + } + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index f91d6e3f..a3d8dab5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,3 +34,6 @@ license-files = ["LICENSE"] [project.urls] Homepage = "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/brendan-w/python-OBD" Issues = "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/brendan-w/python-OBD/issues" + +[project.scripts] +obd-ai-server = "obd_ai.server:main" diff --git a/tests/test_obd_ai_server.py b/tests/test_obd_ai_server.py new file mode 100644 index 00000000..3734312a --- /dev/null +++ b/tests/test_obd_ai_server.py @@ -0,0 +1,124 @@ +import json +import threading +import urllib.request + +import obd +from obd.OBDResponse import OBDResponse +from obd.utils import OBDStatus + +from obd_ai.server import OBDAIFunctionServer, list_tool_definitions, serve_http +from obd_ai.session import OBDAISessionManager +from obd_ai.tools import OBDAIReadOnlyToolSurface + + +class FakeConnection: + def __init__(self, connected=True): + self.connected = connected + self.closed = False + + def query(self, command, force=False): + del force + response = OBDResponse(command=command, messages=[object()]) + response.value = obd.Unit.Quantity(900, obd.Unit.rpm) + return response + + def status(self): + return OBDStatus.CAR_CONNECTED if self.connected else OBDStatus.NOT_CONNECTED + + def is_connected(self): + return self.connected + + @staticmethod + def port_name(): + return "FAKEPORT" + + @staticmethod + def protocol_id(): + return "6" + + @staticmethod + def protocol_name(): + return "ISO 15765-4 (CAN 11/500)" + + @staticmethod + def supports(command): + del command + return True + + def close(self): + self.closed = True + + +class FakeFactory: + def __init__(self, connection): + self.connection = connection + + def __call__(self, **kwargs): + del kwargs + return self.connection + + +def _server_app(): + manager = OBDAISessionManager(obd_factory=FakeFactory(FakeConnection())) + surface = OBDAIReadOnlyToolSurface(session_manager=manager) + return OBDAIFunctionServer(surface=surface) + + +def test_list_tool_definitions_includes_basic_health_scan(): + tools = list_tool_definitions() + names = {tool["name"] for tool in tools} + assert "connect_vehicle" in names + assert "basic_health_scan" in names + + +def test_function_server_calls_tools_with_structured_payloads(): + app = _server_app() + + connected = app.call_tool("connect_vehicle", {}) + payload = app.call_tool("read_sensor", {"command_key": "engine_rpm"}) + + assert connected["ok"] is True + assert payload["ok"] is True + assert payload["data"]["response"]["command"]["name"] == "RPM" + + +def test_function_server_rejects_unknown_tools(): + app = _server_app() + + payload = app.call_tool("clear_trouble_codes", {}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "unknown_tool" + + +def test_http_server_exposes_tools_and_call_endpoint(): + manager = OBDAISessionManager(obd_factory=FakeFactory(FakeConnection())) + surface = OBDAIReadOnlyToolSurface(session_manager=manager) + server = serve_http(host="127.0.0.1", port=0, surface=surface) + thread = threading.Thread(target=server.serve_forever) + thread.daemon = True + thread.start() + + base_url = f"http://127.0.0.1:{server.server_port}" + try: + with urllib.request.urlopen(f"{base_url}/tools") as response: + tools_payload = json.loads(response.read().decode("utf-8")) + + request = urllib.request.Request( + f"{base_url}/call", + data=json.dumps({"tool": "connect_vehicle", "input": {}}).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request) as response: + call_payload = json.loads(response.read().decode("utf-8")) + + assert tools_payload["ok"] is True + assert any(tool["name"] == "read_sensor" for tool in tools_payload["tools"]) + assert call_payload["ok"] is True + assert call_payload["data"]["connection"]["protocol_id"] == "6" + finally: + server.shutdown() + server.server_close() + server.app.close() # type: ignore[attr-defined] + thread.join(timeout=2)