Why Your AI Agent Stack Needs a Gateway, Not a Glue Layer: Lessons from the 11k-Star Orchestration War

Posted on Oct 26 • Originally published at tamiz.pro

The AI agent landscape has shifted rapidly from simple prompt-engineering experiments to complex, multi-step autonomous systems. As these systems grow in sophistication, the underlying architecture that connects LLMs, tools, memory, and workflows becomes the critical differentiator between a fragile prototype and a production-grade platform. The recent

The recent explosion of open-source orchestration frameworks—LangChain, LlamaIndex, AutoGen, CrewAI, and the like—has given developers a false sense of security. They ship a "glue layer": a collection of prompt templates, chain abstractions, and tool wrappers that make the first demo feel magical. But when latency spikes, token costs balloon, or a downstream API deprecates a parameter, the glue dissolves. You're left patching Runnable subclasses at 3 a.m., wondering why your "agent" just sent 40 k tokens to summarize a two-sentence email.

A gateway is not a thicker glue layer. It is a distinct infrastructure tier that sits between your orchestration logic and every external dependency—LLM providers, vector stores, tool APIs, auth systems, observability backends. It enforces contracts, centralizes cross-cutting concerns, and gives you a single lever to pull when the inevitable "provider X changed their response format" incident hits.

The 11k-star project that taught us this lesson the hard way is LangChain. Its early LLMChain and AgentExecutor classes were pure glue. Teams that treated them as infrastructure—baking provider-specific logic into custom chains—spent quarters migrating to LangGraph or LangServe when streaming, human-in-the-loop, and multi-agent handoffs became requirements. The teams that wrapped LangChain behind a thin gateway service from day one swapped providers, added guardrails, and enabled multi-tenancy without touching their business logic.

A gateway needs only three stable interfaces. Everything else is implementation detail.

# gateway/contracts.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import AsyncIterator, Literal @dataclass ( frozen = True ) class ChatMessage: role: Literal [ " system ", " user ", " assistant ", " tool " ] content: str tool_calls: list [ dict ] | None = None tool_call_id: str | None = None @dataclass ( frozen = True ) class ChatRequest: messages: list [ ChatMessage ] model: str | None = None # None → gateway picks default temperature: float = 0.7 max_tokens: int | None = None stream: bool = False metadata: dict | None = None # tenant_id, trace_id, cost_center… @dataclass ( frozen = True ) class ChatResponse: message: ChatMessage usage: dict # prompt_tokens, completion_tokens, cost_usd model: str # actual model used (after routing) latency_ms: int class Gateway ( ABC ): @abstractmethod async def chat ( self, req: ChatRequest ) -> ChatResponse:... @abstractmethod async def chat_stream ( self, req: ChatRequest ) -> AsyncIterator [ ChatResponse ]:... @abstractmethod async def health ( self ) -> bool:... Enter fullscreen mode Exit fullscreen mode That's it. Your orchestration code (LangGraph, custom state machines, whatever) depends only on this ABC.

The gateway implementation handles:

# gateway/impl.py import os, time, uuid, json, logging from contextlib import asynccontextmanager from dataclasses import asdict from typing import AsyncIterator import httpx from tenacity import retry, stop_after_attempt, wait_exponential_jitter from.contracts import ChatMessage, ChatRequest, ChatResponse, Gateway log = logging. getLogger ( " gateway " ) PROVIDER_CONFIG = { " openai ": { " base_url ": " https://api.openai.com/v1 ", " models ": [ " gpt-4o ", " gpt-4o-mini ", " gpt-3.5-turbo " ], " auth ": lambda: { " Authorization ": f " Bearer { os. getenv ( ' OPENAI_API_KEY ' ) } " }, }, " anthropic ": { " base_url ": " https://api.anthropic.com/v1 ", " models ": [ " claude-3-5-sonnet-20241022 ", " claude-3-haiku-20240307 " ], " auth ": lambda: { " x-api-key ": os. getenv ( " ANTHROPIC_API_KEY " )}, }, } class HttpGateway ( Gateway ): def __init__ ( self, default_model: str = " gpt-4o-mini " ): self. default_model = default_model self. _client = httpx. AsyncClient ( timeout = 60.0, limits = httpx. Limits ( max_connections = 100 )) def _pick_provider ( self, model: str | None ) -> tuple [ str, dict ]: model = model or self. default_model for name, cfg in PROVIDER_CONFIG. items (): if model in cfg [ " models " ]: return name, cfg raise ValueError ( f " No provider configured for model { model } " ) @retry ( wait = wait_exponential_jitter ( initial = 1, max = 10 ), stop = stop_after_attempt ( 3 )) async def _post ( self, provider: str, path: str, payload: dict, headers: dict, stream: bool ): cfg = PROVIDER_CONFIG [ provider ] url = f " { cfg [ ' base_url ' ] }{ path } " req_headers = { ** cfg [ " auth " ](), ** headers, " Content-Type ": " application/json " } if stream: return self. _client. stream ( " POST ", url, json = payload, headers = req_headers ) resp = await self. _client. post ( url, json = payload, headers = req_headers ) resp. raise_for_status () return resp. json () def _to_provider_messages ( self, msgs: list [ ChatMessage ]) -> list [ dict ]: out = [] for m in msgs: d = { " role ": m. role, " content ": m. content } if m. tool_calls: d [ " tool_calls " ] = m. tool_calls if m. tool_call_id: d [ " tool_call_id " ] = m. tool_call_id out. append ( d ) return out async def chat ( self, req: ChatRequest ) -> ChatResponse: provider, cfg = self. _pick_provider ( req. model ) start = time. perf_counter () payload = { " model ": req. model or self. default_model, " messages ": self. _to_provider_messages ( req. messages ), " temperature ": req. temperature, " stream ": False, } if req. max_tokens: payload [ " max_tokens " ] = req. max_tokens data = await self. _post ( provider, " /chat/completions ", payload, {}, False ) latency = int (( time. perf_counter () - start ) * 1000 ) choice = data [ " choices " ][ 0 ] msg = ChatMessage ( role = choice [ " message " ][ " role " ], content = choice [ " message " ]. get ( " content ", "" )) usage = data. get ( " usage ", {}) return ChatResponse ( message = msg, usage = { ** usage, " cost_usd ": self. _estimate_cost ( provider, req. model or self. default_model, usage )}, model = data [ " model " ], latency_ms = latency, ) async def chat_stream ( self, req: ChatRequest ) -> AsyncIterator [ ChatResponse ]: provider, cfg = self. _pick_provider ( req. model ) payload = { " model ": req. model or self. default_model, " messages ": self. _to_provider_messages ( req. messages ), " temperature ": req. temperature, " stream ": True, } if req. max_tokens: payload [ " max_tokens " ] = req. max_tokens async with await self. _post ( provider, " /chat/completions ", payload, {}, True ) as resp: async for line in resp. aiter_lines (): if not line. startswith ( " data: " ): continue if line. strip () == " data: [DONE] ": break chunk = json. loads ( line [ 6:]) delta = chunk [ " choices " ][ 0 ]. get ( " delta ", {}) if " content " in delta: yield ChatResponse ( message = ChatMessage ( role = " assistant ", content = delta [ " content " ]), usage = {}, # usage only in final chunk model = chunk. get ( " model ", req. model or self. default_model ), latency_ms = 0, ) def _estimate_cost ( self, provider: str, model: str, usage: dict ) -> float: # Simplified; replace with real pricing table rates = { " gpt-4o ": 0.005, " gpt-4o-mini ": 0.00015, " claude-3-5-sonnet-20241022 ": 0.003 } in_rate = rates. get ( model, 0.001 ) out_rate = in_rate * 3 return ( usage. get ( " prompt_tokens ", 0 ) * in_rate + usage. get ( " completion_tokens ", 0 ) * out_rate ) / 1000 async def health ( self ) -> bool: try: await self. _client. get ( " https://api.openai.com/v1/models ", timeout = 5.0 ) return True except Exception: return False async def aclose ( self ): await self. _client. aclose () Enter fullscreen mode Exit fullscreen mode Key properties:

# orchestrator/agent.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator from gateway.contracts import ChatRequest, ChatMessage, Gateway from gateway.impl import HttpGateway class AgentState ( TypedDict ): messages: Annotated [ list [ ChatMessage ], operator. add ] gateway: Gateway # injected at runtime async def call_model ( state: AgentState ) -> AgentState: gw: Gateway = state [ " gateway " ] req = ChatRequest ( messages = state [ " messages " ], model = " gpt-4o-mini ", temperature = 0.2 ) resp = await gw. chat ( req ) return { " messages ": [ resp. message ]} graph = StateGraph ( AgentState ) graph. add_node ( " llm ", call_model ) graph. set_entry_point ( " llm " ) graph. add_edge ( " llm ", END ) app = graph. compile () # Runtime bootstrap async def run ( user_input: str ): gw = HttpGateway () try: init_state = { " messages ": [ ChatMessage ( role = " user ", content = user_input )], " gateway ": gw } async for event in app. astream ( init_state ): print ( event ) finally: await gw. aclose () Enter fullscreen mode Exit fullscreen mode Notice: the graph knows nothing about OpenAI, Anthropic, retries, or cost. It only knows Gateway. Swap HttpGateway for a LocalLLMGateway (vLLM, Ollama) in staging; prod stays unchanged.

If you're past the prototype phase, the gateway pays for itself the first time a provider drops a parameter or you need to onboard a second LLM vendor.

The orchestration war isn't won by the prettiest chain syntax or the most agents running in parallel. It's won by the team that can swap the model, add a guardrail, or onboard a new tenant on a Friday afternoon without paging the on-call. A gateway gives you that lever. A glue layer gives you a blog post.

Build the gateway. Keep the glue in the application layer where it belongs—thin, testable, and replaceable.