feat!: support credentials per user for tools (with example implementation)

README.md

@@ -78,7 +78,11 @@ tools and merge.
    ```
 
    `TOOLS` is the only required export. Non-empty docstrings are enforced by a
-   contract test.
+   contract test. If a tool does I/O (HTTP, database), write it as
+   `async def` — `@tool` supports coroutines natively; sync tools are fine
+   for pure computation (the runtime runs them in a thread pool). If a tool
+   needs the *user's* credentials, read them from the request headers — see
+   [Per-user credentials](#per-user-credentials).
 
 3. Add tests in `toolsets/my-toolset/tests/test_my_toolset.py` and run
    `./scripts/test`.
@@ -285,6 +289,76 @@ Each Helm release owns its own Ingress for the same host and the controller
 merges them, so the domain's routing table tracks deploys with no central
 config to edit; the index's `/` path only catches what no toolset claims.
 
+### Per-user credentials
+
+Tools that act on a user's behalf (with credentials that differ per calling
+user) must not bake secrets into the deployment — and must not take them as
+tool arguments either, or the model sees them and they land in chat history
+and traces. Instead the client sends them as HTTP headers on every MCP
+call, and the tool reads them at call time:
+
+```python
+from mcp_runtime.credentials import credential_from_header
+
+@tool
+def whoami() -> dict[str, Any]:
+    """Report which account the calling user's credential belongs to."""
+    token = credential_from_header("x-demo-token")
+    ...
+
+TOOLS = [whoami]
+CREDENTIAL_HEADERS = ["x-demo-token"]  # advertised; validated by the contract test
+```
+
+The `CREDENTIAL_HEADERS` export is advertised in the toolset's `/health` and
+in the index's `toolsets` entries, so clients know which toolset needs which
+credential — and send each one *only* to the connections that declare it,
+never to unrelated toolsets. `toolsets/credential-demo` is a working
+(stubbed) example. Clients attach the header per connection — agents by
+decorating the index's `connections` map, `mcp-cli` with `-H`:
+
+```python
+connections = httpx.get("https://<host>/").json()["connections"]
+connections["credential-demo"]["headers"] = {"X-Demo-Token": user_token}
+tools = await MultiServerMCPClient(connections).get_tools()
+```
+
+`mcp-agent` goes further, in the shape a multi-user deployment needs: the
+agent is built **once** and credentials are supplied per call. Each
+connection gets an httpx client factory that, at request time, injects the
+calling user's headers — only those the toolset's advertised declaration
+names (for a direct single-server URL the agent asks the endpoint's sibling
+`/health` for its declaration):
+
+```python
+from mcp_agent.main import user_credentials
+
+with user_credentials({"x-demo-token": the_users_token}):
+    result = await agent.ainvoke(...)
+```
+
+The Chainlit UI builds a settings field (⚙ by the message box) for every
+credential header the connected toolsets advertise and applies the values
+per message — so one long-lived agent process serves many users, each with
+their own credentials.
+
+```sh
+uv run mcp-cli call whoami \
+  --url https://<host>/credential-demo/mcp -H "X-Demo-Token: $TOKEN"
+```
+
+The secret rides the transport (TLS-encrypted at the ingress), never the
+conversation, and the service stays stateless: every call carries its own
+credential, so one pod serves all users. A missing header raises a
+`MissingCredentialError` whose message tells the caller how to supply it.
+Test credential-using tools without a server via
+`mcp_runtime.credentials.header_context`:
+
+```python
+with header_context({"x-demo-token": "secret"}):
+    whoami.invoke({})
+```
+
 ## Development
 
 ```sh

packages/mcp-agent/pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "langchain<2.0.0,>=1.3.7",
     "langchain-mcp-adapters<0.4.0,>=0.3.0",
     "langchain-mistralai<2.0.0,>=1.1.5",
+    "mcp>=1.27.2,<2.0.0",
     "pydantic-settings<3.0.0,>=2.12.0",
     "rich<16.0.0,>=15.0.0",
     "typer<0.27.0,>=0.26.7",

packages/mcp-agent/src/mcp_agent/main.py

@@ -7,11 +7,16 @@
 """
 
 import asyncio
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
 from typing import Annotated, Any, cast
 
 import httpx
 import typer
 from langchain.agents import create_agent
+from mcp.client.streamable_http import create_mcp_http_client
+from mcp.shared.exceptions import McpError
 from langchain_core.messages import BaseMessage, HumanMessage
 from langchain_core.tools import BaseTool
 from langchain_mcp_adapters.client import MultiServerMCPClient
@@ -53,21 +58,164 @@ def connections_from(url: str, payload: Any) -> dict[str, Any]:
     return {"server": {"transport": "streamable_http", "url": url}}
 
 
-def fetch_connections(url: str) -> dict[str, Any]:
-    """Resolve an index or single-server URL to a MultiServerMCPClient config."""
+def credential_headers_from(payload: Any) -> dict[str, list[str]] | None:
+    """Per-toolset credential header names from an index payload.
+
+    ``None`` means the payload was not an index (a direct single-server URL),
+    so no declarations are available.
+    """
+    if not (isinstance(payload, dict) and isinstance(payload.get("toolsets"), list)):
+        return None
+    return {
+        entry["name"]: [
+            header.lower() for header in entry.get("credential_headers", [])
+        ]
+        for entry in payload["toolsets"]
+        if isinstance(entry, dict) and entry.get("name")
+    }
+
+
+# Connection failures an agent should report rather than crash on.
+CONNECT_ERRORS = (httpx.HTTPError, OSError, McpError)
+
+
+def first_leaf(error: BaseException) -> BaseException:
+    """Unwrap (possibly nested) ExceptionGroups to the first real exception."""
+    while isinstance(error, BaseExceptionGroup):
+        error = error.exceptions[0]
+    return error
+
+
+def connect_error_hint(url: str) -> str:
+    """A nudge for the most common misconfiguration: a missing /mcp path."""
+    if url.rstrip("/").endswith("/mcp"):
+        return ""
+    return (
+        " Hint: single-toolset servers serve MCP under /mcp "
+        "(e.g. http://localhost:8000/mcp); only an index is served at the root."
+    )
+
+
+def health_url_for(url: str) -> str | None:
+    """Derive a direct MCP endpoint's sibling /health URL, if there is one."""
+    base = url.rstrip("/")
+    return base.removesuffix("/mcp") + "/health" if base.endswith("/mcp") else None
+
+
+async def single_server_credential_headers(
+    client: httpx.AsyncClient, url: str
+) -> dict[str, list[str]] | None:
+    """Ask a direct MCP endpoint's /health which credential headers it reads.
+
+    Returns ``None`` when there is no health route or it doesn't advertise
+    credentials (e.g. a non-mcp-toolsets server).
+    """
+    health_url = health_url_for(url)
+    if health_url is None:
+        return None
     try:
-        payload = httpx.get(url, follow_redirects=True, timeout=10.0).json()
-    except (httpx.HTTPError, ValueError):
-        payload = None
-    return connections_from(url, payload)
+        health = (await client.get(health_url)).json()
+        headers = health.get("credential_headers")
+    except (httpx.HTTPError, ValueError, AttributeError):
+        return None
+    if not isinstance(headers, list):
+        return None
+    return {"server": [str(header).lower() for header in headers]}
+
+
+async def fetch_connections(
+    url: str,
+) -> tuple[dict[str, Any], dict[str, list[str]] | None]:
+    """Resolve a URL to a MultiServerMCPClient config plus credential needs."""
+    async with httpx.AsyncClient(follow_redirects=True, timeout=10.0) as client:
+        try:
+            payload = (await client.get(url)).json()
+        except (httpx.HTTPError, ValueError):
+            payload = None
+        connections = connections_from(url, payload)
+        required = credential_headers_from(payload)
+        if required is None:
+            required = await single_server_credential_headers(client, url)
+    return connections, required
+
+
+_credentials: ContextVar[dict[str, str] | None] = ContextVar(
+    "user_credentials", default=None
+)
+
+
+@contextmanager
+def user_credentials(headers: dict[str, str] | None) -> Iterator[None]:
+    """Provide the calling user's credential headers for the duration.
+
+    This is how an agent passes a user's secrets to the tools without the
+    model ever seeing them: they ride the MCP transport, not the conversation.
+    The agent is built once; wrap each turn (``run_turn``) in this and the
+    tool calls made inside read the values at request time, so one long-lived
+    agent serves many users with different credentials.
+    """
+    token = _credentials.set(headers)
+    try:
+        yield
+    finally:
+        _credentials.reset(token)
+
+
+def credential_client_factory(allowed: list[str] | None) -> Any:
+    """Build an httpx client factory injecting the current user's credentials.
+
+    Only headers named in ``allowed`` (the toolset's advertised declaration)
+    are injected, so unrelated toolsets never receive them; ``None`` means no
+    declaration was discoverable (a server the user pointed at directly) and
+    every provided header is sent.
+    """
+    wanted = None if allowed is None else {header.lower() for header in allowed}
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        provided = _credentials.get() or {}
+        send = {
+            header: value
+            for header, value in provided.items()
+            if wanted is None or header.lower() in wanted
+        }
+        return create_mcp_http_client(
+            headers={**(headers or {}), **send}, timeout=timeout, auth=auth
+        )
+
+    return factory
+
+
+def with_credential_support(
+    connections: dict[str, Any], required: dict[str, list[str]] | None
+) -> dict[str, Any]:
+    """Wire each connection to inject per-user credentials at call time."""
+    return {
+        name: {
+            **connection,
+            "httpx_client_factory": credential_client_factory(
+                None if required is None else required.get(name, [])
+            ),
+        }
+        for name, connection in connections.items()
+    }
 
 
 async def build_agent(
     url: str, model: str, api_key: SecretStr
 ) -> tuple[Any, dict[str, Any], list[BaseTool]]:
-    """Discover the servers behind ``url`` and build a tool-calling agent."""
-    connections = fetch_connections(url)
-    tools = await MultiServerMCPClient(connections).get_tools()
+    """Discover the servers behind ``url`` and build a tool-calling agent.
+
+    Built once per process/session: per-user credentials are not baked in but
+    read from :func:`user_credentials` on every tool call.
+    """
+    connections, required = await fetch_connections(url)
+    tools = await MultiServerMCPClient(
+        with_credential_support(connections, required)
+    ).get_tools()
     agent = create_agent(
         ChatMistralAI(model_name=model, api_key=api_key),
         tools,
@@ -89,11 +237,13 @@ async def run_turn(
 async def chat_loop(url: str, model: str, api_key: SecretStr) -> None:
     try:
         agent, connections, tools = await build_agent(url, model, api_key)
-    except* (httpx.HTTPError, OSError) as group:
+    except* CONNECT_ERRORS as group:
         console.print(
             f"[red]Could not reach the MCP server(s) behind {url}: "
-            f"{group.exceptions[0]}[/red]"
+            f"{first_leaf(group)}[/red]"
         )
+        if hint := connect_error_hint(url):
+            console.print(f"[yellow]{hint.strip()}[/yellow]")
         raise typer.Exit(1) from None
 
     console.print(