tdfs4ds.agentops.templates¶
The four AgentOps files, as string constants.
They are constants rather than data files on purpose: setup.py's
package_data then needs no new entry, and it matches how
:mod:tdfs4ds.scheduling.script already emits a standalone .py.
Everything awkward about running :func:tdfs4ds.consumer_agent under the
AgentOps runtime is encoded here, because each item is invisible until it
fails in a pod:
The env mapping runs before import tdfs4ds. tdfs4ds/config.py
populates tdfs4ds.<ATTR> exactly once, during import. A variable exported
afterwards never reaches the module config (v0.3.1.155). The pod's environment
is set before the process starts so the ordering is satisfied anyway, but the
fallback assignments below are ours, and they must precede the import.
Setup is at module scope, not in the factory. The platform calls
create_custom_agent() on every invocation. connect() refreshes catalog
views and self-heals missing tables; the Vector Store handshake costs seconds.
Per-invocation that is a broken agent, so the factory returns a thin object over
work already done once per pod.
Turns are serialized. graph._with_domain sets the module-global
tdfs4ds.DATA_DOMAIN and restores it on exit, so two turns in flight in one
process clobber each other's domain — reachable here because the entrypoint is
async and asyncio.to_thread really does run turns concurrently. The lock is
correctness, not tuning; scale with pods.
The thread id never falls back to a constant. session_id if the caller
sent one, else the caller's own identity, else a fresh uuid. A constant default
would put every caller in one shared conversation — answers leaking between
users, which reads as a memory feature until someone notices.
module-attribute
¶
CUSTOMER_AGENT = '"""tdfs4ds consumer agent, packaged for the Teradata AI Studio AgentOps runtime.\n\nGenerated by tdfs4ds {version} - `tdfs4ds.write_agentops_package()`.\nRegenerate rather than hand-editing, so the deployment keeps matching the\nsession it was built from.\n\nDeployment constraints this file depends on (see the fs-deploy-agentops skill):\n\n * replicas: 1 - conversation memory is an in-process MemorySaver, so a\n second replica answers with no history.\n * enable_memory:False- the platform\'s PostgreSQL checkpointer cannot be wired\n into the tdfs4ds graph, so True would provision an\n unused database.\n * one turn at a time - enforced by _TURN_LOCK below. Throughput scales with\n pods, not with in-pod concurrency.\n\nSession memory therefore lives for as long as the pod does, and is lost on\nrestart. That is a known limit, not a misconfiguration.\n"""\n\nimport os\nimport uuid\n\n\ndef _fallback(target, *sources):\n """Set ``target`` from the first non-empty ``sources`` entry, if it is unset.\n\n The AI Studio environment supplies OPENAI_* / TD_* names; tdfs4ds reads\n TDFS4DS_*. setdefault semantics keep whatever config.yaml or runtime_config\n already provided authoritative - the fallback only fills a real gap, and\n only with a value that is actually present (assigning "" would look set and\n fail later, further from the cause).\n """\n if os.environ.get(target):\n return\n for name in sources:\n value = os.environ.get(name)\n if value:\n os.environ[target] = value\n return\n\n\n# -- platform names -> TDFS4DS_* - MUST precede `import tdfs4ds` --------------\n_fallback("TDFS4DS_INSTRUCT_MODEL_API_KEY", "OPENAI_API_KEY")\n_fallback("TDFS4DS_INSTRUCT_MODEL_MODEL", "OPENAI_MODEL_NAME")\n_fallback("TDFS4DS_INSTRUCT_MODEL_URL", "TD_LITELLM_BASE_URL", "TD_AI_STUDIO_AI_MODELHUB")\n_fallback("TDFS4DS_MCP_SERVER_URL", "TD_MCP_SERVER_URL", "MCP_SERVER_URL")\nos.environ.setdefault("TDFS4DS_INSTRUCT_MODEL_PROVIDER", "openai-compatible")\n\nimport asyncio # noqa: E402\nimport logging # noqa: E402\n\nimport tdfs4ds # noqa: E402\nfrom tdfs4ds.agent import consumer_agent # noqa: E402\nfrom tdfs4ds.agentops import session_keeper # noqa: E402\n\nlogger = logging.getLogger("tdfs4ds.agentops")\n\nMCP_ENABLED = {mcp_enabled!r}\n\n# One turn at a time per pod. See the module docstring: this guards a module\n# global, so it is required for correctness even at low traffic.\n_TURN_LOCK = asyncio.Lock()\n\n_READY = False\n\n# Set by the keep-alive when the session stops answering; the next turn then\n# closes it (with a time limit) and opens a new one before it runs.\n_SESSION_DEAD = False\n_KEEPALIVE = None\n\n\ndef _mark_dead():\n global _SESSION_DEAD\n _SESSION_DEAD = True\n\n\ndef _reopen_if_dead():\n """Replace a session the keep-alive found dead. Call inside _TURN_LOCK."""\n global _SESSION_DEAD\n if not _SESSION_DEAD:\n return\n with session_keeper.watchdog("closing the dead Teradata session"):\n session_keeper.close_context_bounded()\n with session_keeper.watchdog("opening a new Teradata session"):\n tdfs4ds.reconnect_teradata(raise_error=True)\n _SESSION_DEAD = False\n\n\ndef _ensure_keepalive():\n """Start the keep-alive on the platform\'s event loop (there is none at import)."""\n global _KEEPALIVE\n if _KEEPALIVE is None or _KEEPALIVE.done():\n _KEEPALIVE = session_keeper.start_keepalive(_TURN_LOCK, _mark_dead)\n\n\ndef _diagnose_connectivity(host, port=1025, timeout=3.0):\n """Say WHY the connection failed: name, route, or credential.\n\n TDML_2000 renders identically for all three, and a pod is not a shell you\n can poke at - so each hypothesis otherwise costs a full redeploy. This runs\n ONLY after a failed connect, so the happy path pays nothing.\n\n A bare name such as "dbc" is completed by the resolver\'s search domains,\n and an AgentOps pod need not share the notebook\'s namespace. Printing those\n domains, and what the name becomes under each, turns "it does not resolve"\n into the fully-qualified name that does.\n """\n import socket\n\n def _resolve(name):\n try:\n infos = socket.getaddrinfo(name, port, proto=socket.IPPROTO_TCP)\n return sorted(set(i[4][0] for i in infos))\n except Exception as exc:\n return "%s: %s" % (type(exc).__name__, exc)\n\n lines = ["connectivity diagnosis for host=%r port=%d" % (host, port)]\n resolved = _resolve(host)\n lines.append(" DNS %-42s -> %s" % (host, resolved))\n\n if isinstance(resolved, list) and resolved:\n try:\n sock = socket.create_connection((resolved[0], port), timeout=timeout)\n sock.close()\n lines.append(" TCP %s:%d reachable -> the name and the route are "\n "fine; the fault is the CREDENTIAL" % (resolved[0], port))\n except Exception as exc:\n lines.append(" TCP %s:%d unreachable -> %s: %s"\n % (resolved[0], port, type(exc).__name__, exc))\n return "\\n".join(lines)\n\n search = []\n try:\n handle = open("/etc/resolv.conf")\n try:\n for line in handle:\n if line.startswith("search ") or line.startswith("domain "):\n search.extend(line.split()[1:])\n finally:\n handle.close()\n except Exception as exc:\n lines.append(" could not read /etc/resolv.conf: %s" % exc)\n\n lines.append(" resolver search domains: %s" % (", ".join(search) or "none"))\n if "." not in host:\n for suffix in search[:6]:\n candidate = "%s.%s" % (host, suffix)\n lines.append(" DNS %-42s -> %s" % (candidate, _resolve(candidate)))\n lines.append(" -> set TDFS4DS_TD_HOST to whichever name resolved above, "\n "fully qualified. If NONE resolved, use the database\'s IP "\n "address: verified working from an AgentOps pod, where DNS "\n "for the database was unavailable entirely.")\n return "\\n".join(lines)\n\n\ndef _bootstrap():\n """Open the connection and prepare the store. Runs once, at import."""\n global _READY\n if _READY:\n return\n\n # Service identity: one long-lived connection from TDFS4DS_TD_*, shared by\n # every caller. reconnect_teradata is reused rather than calling\n # create_context directly, so the pod\'s context is built exactly the way\n # tdfs4ds\'s own auto-reconnect will rebuild it after an idle timeout.\n try:\n tdfs4ds.reconnect_teradata(raise_error=True)\n except Exception:\n # A pod cannot be poked at interactively, so the diagnosis has to be in\n # the log or it costs another deploy cycle.\n logger.error("%s", _diagnose_connectivity(\n os.environ.get("TDFS4DS_TD_HOST", "")))\n raise\n tdfs4ds.connect()\n\n # Printed, not just returned: this lands in the pod log, which is the only\n # diagnostic surface once the container is running.\n try:\n tdfs4ds.check_configuration(display=True)\n except Exception as exc: # a diagnostic must never be what fails startup\n logger.warning("startup configuration check failed: %s", exc)\n\n _READY = True\n\n\n_bootstrap()\n\n\nclass CustomAgent:\n """Thin per-invocation wrapper over the process-wide tdfs4ds agent."""\n\n async def run(self, message, token=None, streaming=False,\n user_id=None, session_id=None, auth_headers=None):\n thread_id = _thread_id(session_id, user_id)\n _ensure_keepalive()\n async with _TURN_LOCK:\n await asyncio.to_thread(_reopen_if_dead)\n _apply_mcp_auth(token, auth_headers)\n # consumer_agent is synchronous and blocks on Teradata and LLM I/O\n # for the whole turn; called directly it would stall the event loop\n # for every other request the pod is serving.\n return await asyncio.to_thread(\n consumer_agent,\n message,\n thread_id=thread_id,\n mcp_enabled=MCP_ENABLED,\n )\n\n\ndef _apply_mcp_auth(token, auth_headers=None):\n """Point MCP at the caller\'s own identity. Call inside _TURN_LOCK.\n\n Prefers the headers the platform SDK built (AuthContext.get_auth_header),\n falling back to formatting the Bearer from the raw token.\n\n tdfs4ds.build_mcp_headers() resolves from module config only - a static\n header dict, or Basic from MCP_USERNAME/MCP_PASSWORD falling back to the\n Teradata password. Neither works in a pod: a static header cannot carry a\n PER-CALLER token, and a JWT environment has no password to derive Basic\n from. The AgentOps runtime hands us the caller\'s token, which is exactly\n what the platform\'s own agent sends to the Teradata MCP server.\n\n Mutating a module global is safe here for the same reason _with_domain\'s\n DATA_DOMAIN mutation is: turns are serialized by _TURN_LOCK.\n """\n if not MCP_ENABLED:\n return\n if auth_headers:\n tdfs4ds.MCP_SERVER_HEADERS = dict(auth_headers)\n elif token:\n tdfs4ds.MCP_SERVER_HEADERS = {{"Authorization": "Bearer %s" % token}}\n\n\ndef _thread_id(session_id, user_id):\n """Conversation key. Never a constant - see the module docstring."""\n if session_id:\n return str(session_id)\n if user_id:\n return "user:%s" % user_id\n # No identity to key on: give this turn its own thread. It gets no history,\n # which is correct - the alternative is joining somebody else\'s.\n return "anon:%s" % uuid.uuid4()\n\n\ndef create_custom_agent() -> CustomAgent:\n """Called per invocation by agent.py. Cheap - setup already happened."""\n return CustomAgent()\n'
module-attribute
¶
AGENT = '"""AgentOps container entry point.\n\nGenerated by tdfs4ds {version}. Platform glue only - the agent logic, and every\ndeployment constraint, lives in customer_agent.py.\n"""\n\nimport sys\nimport os\n\nsys.path.insert(0, os.path.dirname(__file__))\n\nfrom agent_runtime import AgentRuntimeApp, AuthContext\n\nfrom customer_agent import create_custom_agent\n\napp = AgentRuntimeApp()\n\n\n# NOTHING may wrap this function between the decorator and `async def`.\n# AgentRuntimeApp.is_async is `inspect.iscoroutinefunction(self._entrypoint)`,\n# and `_entrypoint` is whatever @app.entrypoint received. Decorators apply\n# bottom-up, so an inner decorator that returns a plain function makes is_async\n# report False and the platform drives an async agent synchronously. Tracing is\n# therefore NOT applied here: the platform collector records the turn, and the\n# node / LLM / tool spans come from the Opik callback tdfs4ds puts in the\n# LangGraph run config when config.yaml sets TRACING: opik (the platform\n# injects the OPIK_* connection settings the Opik SDK reads).\n@app.entrypoint\nasync def invoke(message, streaming=False, session_id=None,\n auth_context: AuthContext = None):\n """One conversational turn.\n\n `streaming` is accepted for signature compatibility and ignored:\n tdfs4ds.consumer_agent_stream is a *synchronous* generator, so bridging it\n to the platform\'s SSE format is not a passthrough.\n """\n agent = create_custom_agent()\n return await agent.run(\n message,\n token=getattr(auth_context, "token", None),\n # The SDK builds this itself; AuthContext.get_auth_header() is the\n # documented way to authenticate downstream calls (its own docstring\n # names the MCP server), so it is used rather than formatting the\n # header by hand.\n auth_headers=(auth_context.get_auth_header() if auth_context else None),\n streaming=False,\n user_id=getattr(auth_context, "user_id", None),\n session_id=session_id,\n )\n'
module-attribute
¶
REQUIREMENTS = "# Generated by tdfs4ds {version}.\n#\n# tdfs4ds pulls a broad dependency set (teradataml, langchain, chromadb, plotly,\n# networkx, gradio). Size the pod accordingly - 2Gi / 1000m - rather than\n# trimming this file: the base install is not currently separable into extras.\ntdfs4ds=={version}\n# Traces the agent's graph nodes and LLM/tool calls (config.yaml: TRACING: opik).\nopik\n"
module-attribute
¶
CONFIG_HEADER = '# Generated by tdfs4ds {version} from the running session.\n#\n# Bundled in the deployment zip and COMMITTED - so it carries only settings that\n# are safe to read. Credentials and connection details are deliberately absent;\n# supply them through `runtime_config` in the deployment payload, which the\n# platform injects into the pod environment at deploy time.\n#\n# Regenerate with tdfs4ds.write_agentops_package() rather than editing, so this\n# keeps matching the session it was built from.\nenvironment_variables:\n'
module-attribute
¶
CUSTOMER_AGENT_JWT = '"""tdfs4ds consumer agent - AgentOps runtime, per-caller JWT identity.\n\nGenerated by tdfs4ds {version} - `tdfs4ds.write_agentops_package(auth="jwt")`.\nRegenerate rather than hand-editing.\n\nEach caller\'s own token opens the Teradata session, so feature-store reads carry\nthe caller\'s identity and authorization. Consequences worth knowing:\n\n * every caller needs SELECT rights on the feature store - there is no service\n account behind this;\n * switching caller costs a reconnect (the context is process-wide);\n * a token expiring mid-conversation is self-healing: the next call brings a\n fresh one and the context is rebuilt.\n\nDeployment constraints (see the fs-deploy-agentops skill):\n\n * replicas: 1 - conversation memory is an in-process MemorySaver.\n * enable_memory:False- the platform checkpointer cannot be wired into the\n tdfs4ds graph.\n * one turn at a time - enforced by _TURN_LOCK. Required here for a second\n reason: the teradataml context is process-wide, so\n concurrent turns would run under the wrong identity.\n"""\n\nimport os\nimport uuid\n\n\ndef _fallback(target, *sources):\n """Set ``target`` from the first non-empty ``sources`` entry, if it is unset."""\n if os.environ.get(target):\n return\n for name in sources:\n value = os.environ.get(name)\n if value:\n os.environ[target] = value\n return\n\n\n# -- platform names -> TDFS4DS_* - MUST precede `import tdfs4ds` --------------\n_fallback("TDFS4DS_INSTRUCT_MODEL_API_KEY", "OPENAI_API_KEY")\n_fallback("TDFS4DS_INSTRUCT_MODEL_MODEL", "OPENAI_MODEL_NAME")\n_fallback("TDFS4DS_INSTRUCT_MODEL_URL", "TD_LITELLM_BASE_URL", "TD_AI_STUDIO_AI_MODELHUB")\n_fallback("TDFS4DS_MCP_SERVER_URL", "TD_MCP_SERVER_URL", "MCP_SERVER_URL")\nos.environ.setdefault("TDFS4DS_INSTRUCT_MODEL_PROVIDER", "openai-compatible")\nos.environ.setdefault("TDFS4DS_TD_HOST", "dbc")\nos.environ.setdefault("TDFS4DS_TD_LOGMECH", "JWT")\n\nimport asyncio # noqa: E402\nimport hashlib # noqa: E402\nimport logging # noqa: E402\n\nimport tdfs4ds # noqa: E402\nfrom tdfs4ds.agent import consumer_agent # noqa: E402\nfrom tdfs4ds.agentops import session_keeper # noqa: E402\n\nlogger = logging.getLogger("tdfs4ds.agentops")\n\nMCP_ENABLED = {mcp_enabled!r}\n\n# Serializes turns. Required for correctness twice over: tdfs4ds mutates the\n# module global DATA_DOMAIN per turn, AND the teradataml context below is\n# process-wide, so an interleaved turn would query as the wrong caller.\n_TURN_LOCK = asyncio.Lock()\n\n# Fingerprint of the token the current context was built from. A hash, not the\n# token: this lives for the life of the process and never needs to be readable.\n_CONNECTED_AS = None\n_STORE_READY = False\n_KEEPALIVE = None\n\n\ndef _mark_dead():\n """The keep-alive found the session dead: the next turn opens a new one."""\n global _CONNECTED_AS\n _CONNECTED_AS = None\n\n\ndef _ensure_keepalive():\n """Start the keep-alive on the platform\'s event loop (there is none at import)."""\n global _KEEPALIVE\n if _KEEPALIVE is None or _KEEPALIVE.done():\n _KEEPALIVE = session_keeper.start_keepalive(_TURN_LOCK, _mark_dead)\n\n\ndef _fingerprint(token):\n return hashlib.sha256((token or "").encode("utf-8")).hexdigest()[:16]\n\n\ndef _diagnose_connectivity(host, port=1025, timeout=3.0):\n """Say WHY the connection failed: name, route, or credential.\n\n TDML_2000 renders identically for all three, and a pod is not a shell you\n can poke at - so each hypothesis otherwise costs a full redeploy. This runs\n ONLY after a failed connect, so the happy path pays nothing.\n\n A bare name such as "dbc" is completed by the resolver\'s search domains,\n and an AgentOps pod need not share the notebook\'s namespace. Printing those\n domains, and what the name becomes under each, turns "it does not resolve"\n into the fully-qualified name that does.\n """\n import socket\n\n def _resolve(name):\n try:\n infos = socket.getaddrinfo(name, port, proto=socket.IPPROTO_TCP)\n return sorted(set(i[4][0] for i in infos))\n except Exception as exc:\n return "%s: %s" % (type(exc).__name__, exc)\n\n lines = ["connectivity diagnosis for host=%r port=%d" % (host, port)]\n resolved = _resolve(host)\n lines.append(" DNS %-42s -> %s" % (host, resolved))\n\n if isinstance(resolved, list) and resolved:\n try:\n sock = socket.create_connection((resolved[0], port), timeout=timeout)\n sock.close()\n lines.append(" TCP %s:%d reachable -> the name and the route are "\n "fine; the fault is the CREDENTIAL" % (resolved[0], port))\n except Exception as exc:\n lines.append(" TCP %s:%d unreachable -> %s: %s"\n % (resolved[0], port, type(exc).__name__, exc))\n return "\\n".join(lines)\n\n search = []\n try:\n handle = open("/etc/resolv.conf")\n try:\n for line in handle:\n if line.startswith("search ") or line.startswith("domain "):\n search.extend(line.split()[1:])\n finally:\n handle.close()\n except Exception as exc:\n lines.append(" could not read /etc/resolv.conf: %s" % exc)\n\n lines.append(" resolver search domains: %s" % (", ".join(search) or "none"))\n if "." not in host:\n for suffix in search[:6]:\n candidate = "%s.%s" % (host, suffix)\n lines.append(" DNS %-42s -> %s" % (candidate, _resolve(candidate)))\n lines.append(" -> set TDFS4DS_TD_HOST to whichever name resolved above, "\n "fully qualified.")\n return "\\n".join(lines)\n\n\ndef _ensure_connected(token):\n """Make the process-wide context belong to *token*. Call inside _TURN_LOCK."""\n global _CONNECTED_AS, _STORE_READY\n\n if not token:\n raise RuntimeError(\n "No caller token. This agent authenticates to Teradata with the "\n "caller\'s own JWT (auth_context.token); it has no service account "\n "to fall back on.")\n\n fingerprint = _fingerprint(token)\n if fingerprint == _CONNECTED_AS:\n return\n\n # reconnect_teradata reads TDFS4DS_TD_*, so the token is handed over the\n # same way every other credential is - one code path, and it means the\n # automatic reconnect on a dropped session works here too.\n os.environ["TDFS4DS_TD_LOGDATA"] = "token=%s" % token\n logger.info("opening Teradata session for caller %s", fingerprint)\n # Closing the previous session can hang on a connection dropped while idle\n # (seen live, 0.3.6.0): it gets a time limit, then is abandoned.\n with session_keeper.watchdog("closing the previous Teradata session"):\n session_keeper.close_context_bounded()\n try:\n with session_keeper.watchdog("opening the Teradata session"):\n tdfs4ds.reconnect_teradata(raise_error=True)\n except Exception:\n logger.error("%s", _diagnose_connectivity(\n os.environ.get("TDFS4DS_TD_HOST", "")))\n raise\n _CONNECTED_AS = fingerprint\n\n if not _STORE_READY:\n # Catalog views belong to the STORE, not the caller, so this is done\n # once per process rather than on every identity switch.\n tdfs4ds.connect()\n try:\n tdfs4ds.check_configuration(display=True)\n except Exception as exc: # a diagnostic must never break startup\n logger.warning("configuration check failed: %s", exc)\n _STORE_READY = True\n\n\nclass CustomAgent:\n """Thin per-invocation wrapper over the process-wide tdfs4ds agent."""\n\n async def run(self, message, token=None, streaming=False,\n user_id=None, session_id=None, auth_headers=None):\n thread_id = _thread_id(session_id, user_id)\n _ensure_keepalive()\n async with _TURN_LOCK:\n await asyncio.to_thread(_ensure_connected, token)\n _apply_mcp_auth(token, auth_headers)\n # consumer_agent is synchronous and blocks on Teradata and LLM I/O.\n return await asyncio.to_thread(\n consumer_agent,\n message,\n thread_id=thread_id,\n mcp_enabled=MCP_ENABLED,\n )\n\n\ndef _apply_mcp_auth(token, auth_headers=None):\n """Point MCP at the caller\'s own identity. Call inside _TURN_LOCK.\n\n Prefers the headers the platform SDK built (AuthContext.get_auth_header),\n falling back to formatting the Bearer from the raw token.\n\n tdfs4ds.build_mcp_headers() resolves from module config only - a static\n header dict, or Basic from MCP_USERNAME/MCP_PASSWORD falling back to the\n Teradata password. Neither works in a pod: a static header cannot carry a\n PER-CALLER token, and a JWT environment has no password to derive Basic\n from. The AgentOps runtime hands us the caller\'s token, which is exactly\n what the platform\'s own agent sends to the Teradata MCP server.\n\n Mutating a module global is safe here for the same reason _with_domain\'s\n DATA_DOMAIN mutation is: turns are serialized by _TURN_LOCK.\n """\n if not MCP_ENABLED:\n return\n if auth_headers:\n tdfs4ds.MCP_SERVER_HEADERS = dict(auth_headers)\n elif token:\n tdfs4ds.MCP_SERVER_HEADERS = {{"Authorization": "Bearer %s" % token}}\n\n\ndef _thread_id(session_id, user_id):\n """Conversation key. Never a constant - a constant shares one conversation\n between every caller."""\n if session_id:\n return str(session_id)\n if user_id:\n return "user:%s" % user_id\n return "anon:%s" % uuid.uuid4()\n\n\ndef create_custom_agent() -> CustomAgent:\n """Called per invocation by agent.py. Cheap - the connection is lazy."""\n return CustomAgent()\n'
module-attribute
¶
AGENT_RUNTIME_INIT = '"""agent_runtime - AgentOps runtime contract.\n\nGenerated by tdfs4ds {version} because no platform-supplied agent_runtime/ was\npresent. If AgentOps ships one, copy it into this directory and it will be used\ninstead - a platform-supplied copy is never overwritten.\n"""\n\nfrom .runtime import AgentRuntimeApp, AuthContext, agent_runtime_app\n\n__all__ = ["AgentRuntimeApp", "AuthContext", "agent_runtime_app"]\n'
module-attribute
¶
AGENT_RUNTIME_RUNTIME = '"""Compatibility implementation of the AgentOps runtime contract.\n\nGenerated by tdfs4ds {version}. Two symbols the platform requires an agent\nmodule to expose:\n\n``AuthContext``\n The caller\'s verified identity, passed to the entrypoint. ``token`` is the\n raw JWT used for downstream calls (the Teradata MCP server, AI Studio);\n ``get_auth_header()`` is the supported way to build the header for one.\n\n``AgentRuntimeApp``\n Holds the single registered entrypoint. The platform host reads\n ``is_async`` to decide how to drive it, so ``entrypoint`` registers the\n function it is given **unchanged** - a decorator applied beneath it becomes\n what is registered, and one returning a plain function would make an async\n agent look synchronous.\n"""\n\nfrom __future__ import annotations\n\nimport inspect\nfrom collections.abc import Callable\nfrom dataclasses import dataclass, field\nfrom typing import Any\n\n\n@dataclass(frozen=True)\nclass AuthContext:\n """Authenticated caller identity, supplied per invocation.\n\n Frozen: a turn must not be able to alter the identity it runs under.\n """\n\n token: str = ""\n user_id: str = ""\n email: str = ""\n customer_id: str = ""\n roles: tuple = field(default_factory=tuple)\n claims: dict[str, Any] = field(default_factory=dict)\n\n def get_auth_header(self) -> dict[str, str]:\n """Authorization header for a downstream call, or ``{{}}`` with no token."""\n if self.token:\n return {{"Authorization": "Bearer %s" % self.token}}\n return {{}}\n\n def has_role(self, role: str) -> bool:\n """True when the caller holds *role*."""\n return role in self.roles\n\n def has_any_role(self, roles: list) -> bool:\n """True when the caller holds any of *roles*."""\n return any(role in self.roles for role in roles)\n\n\nclass AgentRuntimeApp:\n """Container for the agent entrypoint."""\n\n def __init__(self):\n self._entrypoint: Callable[..., Any] | None = None\n\n def entrypoint(self, func: Callable[..., Any]):\n """Register *func* as the handler. One per application."""\n if self._entrypoint is not None:\n raise RuntimeError("An entrypoint is already registered")\n self._entrypoint = func\n\n def wrapper(*args, **kwargs):\n return func(*args, **kwargs)\n\n self.invoke = wrapper\n return wrapper\n\n @property\n def is_async(self) -> bool:\n """Whether the registered entrypoint is a coroutine function."""\n return (inspect.iscoroutinefunction(self._entrypoint)\n if self._entrypoint else False)\n\n\ndef agent_runtime_app() -> AgentRuntimeApp:\n """Factory for a fresh application."""\n return AgentRuntimeApp()\n'