feat(runtime): add credential resolver and request context for SGP delegation

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: cdvillegas

src/agentex/lib/runtime/__init__.py

@@ -0,0 +1,36 @@
+from agentex.lib.runtime.models import Credentials, CredentialScheme
+from agentex.lib.runtime.context import (
+    RequestContext,
+    current_request,
+    request_context_scope,
+    get_credential_resolver,
+    set_credential_resolver,
+    run_with_request_context,
+    wrap_async_generator_with_request_context,
+)
+from agentex.lib.runtime.resolver import (
+    SGP_TARGET,
+    ENV_SGP_API_KEY,
+    HEADER_ACTING_AS_AGENT,
+    HEADER_ACTING_USER_API_KEY,
+    CredentialResolver,
+    PassthroughResolver,
+)
+
+__all__ = [
+    "CredentialResolver",
+    "CredentialScheme",
+    "Credentials",
+    "ENV_SGP_API_KEY",
+    "HEADER_ACTING_AS_AGENT",
+    "HEADER_ACTING_USER_API_KEY",
+    "PassthroughResolver",
+    "RequestContext",
+    "SGP_TARGET",
+    "current_request",
+    "get_credential_resolver",
+    "request_context_scope",
+    "run_with_request_context",
+    "set_credential_resolver",
+    "wrap_async_generator_with_request_context",
+]

src/agentex/lib/runtime/context.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import contextvars
+from typing import TypeVar
+from contextlib import asynccontextmanager
+from collections.abc import Callable, Awaitable, AsyncIterator, AsyncGenerator
+
+from agentex.lib.runtime.models import Credentials
+from agentex.lib.runtime.resolver import CredentialResolver, PassthroughResolver
+
+T = TypeVar("T")
+
+_ctx_var_request_context: contextvars.ContextVar[RequestContext | None] = contextvars.ContextVar(
+    "agentex_request_context", default=None
+)
+
+_default_resolver: CredentialResolver = PassthroughResolver()
+
+
+def set_credential_resolver(resolver: CredentialResolver) -> None:
+    """Configure the credential resolver used by runtime request context."""
+    global _default_resolver
+    _default_resolver = resolver
+
+
+def get_credential_resolver() -> CredentialResolver:
+    return _default_resolver
+
+
+class RequestContext:
+    """Per-request runtime context populated by FastACP on each inbound RPC."""
+
+    def __init__(
+        self,
+        *,
+        headers: dict[str, str],
+        agent_id: str,
+        resolver: CredentialResolver,
+    ) -> None:
+        self._headers = headers
+        self._agent_id = agent_id
+        self._resolver = resolver
+
+    @classmethod
+    def from_headers(
+        cls,
+        headers: dict[str, str],
+        agent_id: str,
+        resolver: CredentialResolver | None = None,
+    ) -> RequestContext:
+        return cls(
+            headers=headers,
+            agent_id=agent_id,
+            resolver=resolver or get_credential_resolver(),
+        )
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return self._headers
+
+    @property
+    def agent_id(self) -> str:
+        return self._agent_id
+
+    async def get_credentials_for(self, target: str) -> Credentials:
+        return await self._resolver.resolve(self._headers, self._agent_id, target)
+
+    async def get_token(self, target: str = "sgp") -> str:
+        """Return the raw credential value for a target (convenience wrapper)."""
+        credentials = await self.get_credentials_for(target)
+        return credentials.value
+
+
+def current_request() -> RequestContext:
+    """Return the active request context for the current RPC handler."""
+    context = _ctx_var_request_context.get()
+    if context is None:
+        raise RuntimeError(
+            "No active Agentex request context. Call current_request() only "
+            "from code running inside an agent RPC handler."
+        )
+    return context
+
+
+@asynccontextmanager
+async def request_context_scope(
+    headers: dict[str, str],
+    agent_id: str,
+    resolver: CredentialResolver | None = None,
+) -> AsyncGenerator[RequestContext, None]:
+    context = RequestContext.from_headers(headers, agent_id, resolver)
+    token = _ctx_var_request_context.set(context)
+    try:
+        yield context
+    finally:
+        _ctx_var_request_context.reset(token)
+
+
+async def run_with_request_context(
+    headers: dict[str, str],
+    agent_id: str,
+    fn: Callable[[], Awaitable[T]],
+    *,
+    resolver: CredentialResolver | None = None,
+) -> T:
+    async with request_context_scope(headers, agent_id, resolver):
+        return await fn()
+
+
+async def wrap_async_generator_with_request_context(
+    async_gen: AsyncIterator[T],
+    headers: dict[str, str],
+    agent_id: str,
+    *,
+    resolver: CredentialResolver | None = None,
+) -> AsyncGenerator[T, None]:
+    """Keep request context active while a streaming handler yields chunks."""
+    async with request_context_scope(headers, agent_id, resolver):
+        async for item in async_gen:
+            yield item

src/agentex/lib/runtime/models.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import Field, BaseModel
+
+CredentialScheme = Literal["api_key", "bearer"]
+
+
+class Credentials(BaseModel):
+    """Resolved outbound credentials for a downstream target."""
+
+    scheme: CredentialScheme = Field(
+        ...,
+        description="How to attach the credential to an HTTP request",
+    )
+    value: str = Field(..., description="Secret credential value")

src/agentex/lib/runtime/resolver.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import os
+from typing import Protocol, runtime_checkable
+
+from agentex.lib.runtime.models import Credentials
+
+HEADER_ACTING_USER_API_KEY = "x-acting-user-api-key"
+HEADER_ACTING_AS_AGENT = "x-acting-as-agent"
+SGP_TARGET = "sgp"
+ENV_SGP_API_KEY = "SGP_API_KEY"
+
+
+def _normalize_headers(headers: dict[str, str]) -> dict[str, str]:
+    return {key.lower(): value for key, value in headers.items()}
+
+
+@runtime_checkable
+class CredentialResolver(Protocol):
+    """Plugin interface for resolving outbound credentials by target name."""
+
+    async def resolve(
+        self,
+        headers: dict[str, str],
+        agent_id: str,
+        target: str,
+    ) -> Credentials:
+        """Return credentials for the given target using inbound request context."""
+        ...
+
+
+class PassthroughResolver:
+    """Default resolver: forwarded user API key, then legacy SGP_API_KEY fallback."""
+
+    async def resolve(
+        self,
+        headers: dict[str, str],
+        agent_id: str,
+        target: str,
+    ) -> Credentials:
+        del agent_id  # reserved for future per-agent resolution (OBO, vault lookups)
+
+        normalized = _normalize_headers(headers)
+        api_key = normalized.get(HEADER_ACTING_USER_API_KEY) or os.environ.get(ENV_SGP_API_KEY)
+        if not api_key:
+            raise RuntimeError(
+                "No credential available for target "
+                f"'{target}': expected inbound header "
+                f"'{HEADER_ACTING_USER_API_KEY}' or environment variable "
+                f"'{ENV_SGP_API_KEY}'."
+            )
+
+        if target == SGP_TARGET:
+            return Credentials(scheme="api_key", value=api_key)
+
+        # v1 passthrough uses the same user API key shape for all targets.
+        return Credentials(scheme="bearer", value=api_key)