diff --git a/Makefile b/Makefile
index c491b81..8e3a594 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: help setup run test test-unit test-integration lint typecheck clean
+.PHONY: help setup run test test-unit test-integration test-contracts lint typecheck clean
 
 VENV := .venv
 PYTHON := $(VENV)/bin/python
@@ -29,13 +29,16 @@ run: ## Start the development server on port 8777
 	$(UVICORN) src.main:app --reload --host $(HOST) --port $(PORT) 2>&1 | tee build/logs/server.log
 
 test: ## Run unit tests (no external services needed)
-	$(PYTEST) --ignore=tests/e2e --ignore=tests/integration
+	$(PYTEST) --ignore=tests/e2e --ignore=tests/integration --ignore=tests/contracts
 
 test-unit: test ## Alias for test
 
 test-integration: ## Run integration tests (needs Claude/Ollama)
 	$(PYTEST) tests/agents/test_tatlock_agent.py -v
 
+test-contracts: ## Wire-level contract tests against live service boundaries
+	$(PYTEST) tests/contracts -v --no-cov
+
 lint: ## Run ruff linter and formatter check
 	$(RUFF) check src tests
 	$(RUFF) format --check src tests
diff --git a/pyproject.toml b/pyproject.toml
index e14b464..3695c39 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -47,6 +47,7 @@ markers = [
     "unit: Unit tests",
     "integration: Integration tests",
     "slow: Slow running tests",
+    "contract: Wire-level contract tests against live service boundaries",
 ]
 addopts = [
     "--verbose",
diff --git a/tests/contracts/__init__.py b/tests/contracts/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/contracts/test_service_contracts.py b/tests/contracts/test_service_contracts.py
new file mode 100644
index 0000000..4cadd8e
--- /dev/null
+++ b/tests/contracts/test_service_contracts.py
@@ -0,0 +1,219 @@
+"""
+Wire-level contract tests for external service boundaries.
+
+Each test sends the raw request the application code sends (no client
+wrappers, no mocks) and asserts on the response shape, so boundary
+breakage is caught directly instead of surfacing as agent misbehavior.
+
+Semantics:
+- Service unreachable  -> skip (an outage is not a contract violation)
+- Service reachable but wrong response shape -> fail
+
+Run with: make test-contracts
+"""
+import json
+
+import httpx
+import pytest
+
+from src.core.config import config
+
+OLLAMA = str(config.OLLAMA_HOST).rstrip("/")
+QDRANT = f"http://{config.QDRANT_HOST}:{config.QDRANT_PORT}"
+SEARXNG = str(config.SEARXNG_HOST).rstrip("/")
+
+CALCULATOR_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "calculator",
+        "description": "Evaluate a math expression",
+        "parameters": {
+            "type": "object",
+            "properties": {"expression": {"type": "string"}},
+            "required": ["expression"],
+        },
+    },
+}
+
+
+async def _get_or_skip(url: str, service: str, timeout: float = 5.0) -> httpx.Response:
+    """GET a URL, skipping the test if the service is unreachable."""
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            return await client.get(url)
+    except httpx.TransportError as e:
+        pytest.skip(f"{service} unreachable at {url}: {e}")
+
+
+async def _post_or_skip(
+    url: str, service: str, payload: dict, timeout: float, headers: dict | None = None
+) -> httpx.Response:
+    """POST a payload, skipping the test if the service is unreachable."""
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            return await client.post(url, json=payload, headers=headers)
+    except httpx.TransportError as e:
+        pytest.skip(f"{service} unreachable at {url}: {e}")
+
+
+@pytest.mark.contract
+class TestOllamaContract:
+    """Boundary: Ollama native API and its OpenAI-compat layer."""
+
+    async def test_tags_lists_configured_model(self):
+        # Mirrors check_ollama_health()
+        response = await _get_or_skip(f"{OLLAMA}/api/tags", "ollama")
+        assert response.status_code == 200
+        names = [m["name"] for m in response.json()["models"]]
+        model = config.OLLAMA_DEFAULT_MODEL
+        assert model in names or f"{model}:latest" in names, (
+            f"{model} not pulled; available: {names}"
+        )
+
+    async def test_generate_returns_plain_text(self):
+        # Mirrors StewardAgent._call_ollama()
+        response = await _post_or_skip(
+            f"{OLLAMA}/api/generate",
+            "ollama",
+            {
+                "model": config.OLLAMA_DEFAULT_MODEL,
+                "prompt": "Reply with the single word: pong",
+                "stream": False,
+                "options": {"temperature": 0.3, "top_p": 0.9},
+            },
+            timeout=config.OLLAMA_TIMEOUT,
+        )
+        assert response.status_code == 200
+        assert response.json()["response"].strip()
+
+    async def test_openai_compat_tool_calling(self):
+        # Mirrors the request PydanticAI's OpenAIChatModel sends for the
+        # orchestration phase, including the extra_body tool_choice.
+        response = await _post_or_skip(
+            f"{OLLAMA}/v1/chat/completions",
+            "ollama",
+            {
+                "model": config.OLLAMA_DEFAULT_MODEL,
+                "messages": [
+                    {"role": "user", "content": "What is 6 * 7? Use the calculator."}
+                ],
+                "tools": [CALCULATOR_TOOL],
+                "tool_choice": "required",
+                "stream": False,
+            },
+            timeout=config.OLLAMA_TIMEOUT,
+        )
+        assert response.status_code == 200
+        message = response.json()["choices"][0]["message"]
+        tool_calls = message.get("tool_calls")
+        assert tool_calls, f"model answered in text instead of calling the tool: {message}"
+        assert tool_calls[0]["function"]["name"] == "calculator"
+        arguments = json.loads(tool_calls[0]["function"]["arguments"])
+        assert "expression" in arguments
+
+
+@pytest.mark.contract
+class TestAnthropicContract:
+    """Boundary: Anthropic Messages API (the Claude fallback backend)."""
+
+    HEADERS_KEY = "anthropic-version"
+
+    def _headers(self) -> dict:
+        if not config.ANTHROPIC_API_KEY:
+            pytest.skip("ANTHROPIC_API_KEY not configured")
+        return {
+            "x-api-key": config.ANTHROPIC_API_KEY,
+            "anthropic-version": "2023-06-01",
+        }
+
+    async def test_minimal_message_accepted(self):
+        # Mirrors check_claude_health(): tiny request, no sampling params
+        response = await _post_or_skip(
+            "https://api.anthropic.com/v1/messages",
+            "anthropic",
+            {
+                "model": config.ANTHROPIC_MODEL,
+                "max_tokens": 1,
+                "messages": [{"role": "user", "content": "hi"}],
+            },
+            timeout=30.0,
+            headers=self._headers(),
+        )
+        assert response.status_code == 200, response.text
+
+    async def test_temperature_rejected(self):
+        # Pins the Claude Sonnet 5+ contract that broke the Steward:
+        # sampling parameters are rejected with a 400 (and not billed).
+        response = await _post_or_skip(
+            "https://api.anthropic.com/v1/messages",
+            "anthropic",
+            {
+                "model": config.ANTHROPIC_MODEL,
+                "max_tokens": 1,
+                "messages": [{"role": "user", "content": "hi"}],
+                "temperature": 0.3,
+            },
+            timeout=30.0,
+            headers=self._headers(),
+        )
+        assert response.status_code == 400
+        assert "temperature" in response.text
+
+
+@pytest.mark.contract
+class TestQdrantContract:
+    """Boundary: Qdrant REST API (Biographer's vector memory)."""
+
+    async def test_collections_endpoint(self):
+        response = await _get_or_skip(f"{QDRANT}/collections", "qdrant")
+        assert response.status_code == 200
+        assert "collections" in response.json()["result"]
+
+
+@pytest.mark.contract
+class TestSearxngContract:
+    """Boundary: SearXNG JSON search API (web search tool)."""
+
+    async def test_json_search(self):
+        response = await _get_or_skip(
+            f"{SEARXNG}/search?q=test&format=json", "searxng", timeout=config.SEARXNG_TIMEOUT
+        )
+        assert response.status_code == 200
+        assert "results" in response.json()
+
+
+@pytest.mark.contract
+class TestLibraryDeskContract:
+    """Boundary: library-desk research API (the Librarian's backend)."""
+
+    async def test_health(self):
+        host = getattr(config, "LIBRARY_DESK_HOST", None)
+        if not host:
+            pytest.skip("LIBRARY_DESK_HOST not configured")
+        response = await _get_or_skip(f"{str(host).rstrip('/')}/health", "library-desk")
+        assert response.status_code == 200
+
+
+@pytest.mark.contract
+class TestRedisContract:
+    """Boundary: Redis on the configured memory DB."""
+
+    async def test_roundtrip(self):
+        import redis.asyncio as redis
+
+        client = redis.Redis(
+            host=config.REDIS_HOST,
+            port=config.REDIS_PORT,
+            db=config.REDIS_MEMORY_DB,
+            socket_connect_timeout=3,
+        )
+        try:
+            await client.ping()
+        except Exception as e:
+            pytest.skip(f"redis unreachable: {e}")
+        try:
+            await client.set("contract-test-key", "ok", ex=30)
+            assert await client.get("contract-test-key") == b"ok"
+            await client.delete("contract-test-key")
+        finally:
+            await client.aclose()