langgraph-responses-gateway 0.2.0

Expose any LangGraph agent as an OpenAI Responses API endpoint

LangGraph Responses Gateway

Bridge any LangGraph agent to OpenAI's Responses API format with zero configuration.

Overview

langgraph-responses-gateway is a lightweight Python package that exposes any LangGraph agent or multi-agent system as an OpenAI Responses API endpoint. This enables seamless integration with modern AI clients like Vercel AI SDK, OpenAI's client libraries, and any tool that supports the Responses API format.

Why This Package?

The AI ecosystem has a gap: while LangGraph provides powerful agent orchestration capabilities, and OpenAI's Responses API has become the de facto standard for AI interactions, there's no simple way to bridge the two. This package fills that gap.

Key Benefits

• Zero Configuration: Works with any LangGraph CompiledGraph out of the box
• Full Streaming Support: Native SSE streaming with correct OpenAI event names
• Conversation Chaining: Support for previous_response_id to maintain context
• OpenAI Spec Compliant: Uses input parameter and exact event structure
• Automatic Translation: Converts OpenAI input to LangGraph messages format
• Vercel AI SDK Compatible: Seamless integration with modern web frameworks
• Production Ready: Built on FastAPI with robust error handling

Installation

pip install langgraph-responses-gateway

Or with uv:

uv add langgraph-responses-gateway

Quick Start

Transform your LangGraph agent into an OpenAI-compatible API in just 3 lines:

from langgraph_responses_gateway import ResponsesGateway
from your_agent import create_agent_graph

# 1. Create your LangGraph
graph = create_agent_graph()

# 2. Wrap it as Responses API
gateway = ResponsesGateway(graph, name="My Agent")

# 3. Run the server
import uvicorn
uvicorn.run(gateway.app, host="0.0.0.0", port=8000)

Your agent is now accessible at http://localhost:8000/v1/responses!

Usage Examples

Basic Chat Request

import httpx

# Using 'input' (OpenAI Responses API spec)
response = httpx.post(
    "http://localhost:8000/v1/responses",
    json={
        "model": "langgraph-agent",  # Required
        "input": "Hello!",
        "stream": False
    }
)
print(response.json()["output"][0]["content"][0]["text"])

Streaming with SSE

import httpx
import json

with httpx.stream("POST", "http://localhost:8000/v1/responses",
                  json={
                      "model": "langgraph-agent",
                      "input": "Tell me a story",
                      "stream": True
                  }) as r:
    for line in r.iter_lines():
        if line.startswith("data: "):
            event = json.loads(line[6:])
            if event["type"] == "response.output_text.delta":
                print(event["delta"], end="", flush=True)

Conversation Chaining

# First message
response1 = httpx.post(
    "http://localhost:8000/v1/responses",
    json={
        "model": "langgraph-agent",
        "input": "My name is Alice",
        "store": True,  # Store for chaining
        "stream": False
    }
)
response_id = response1.json()["id"]

# Chained follow-up
response2 = httpx.post(
    "http://localhost:8000/v1/responses",
    json={
        "model": "langgraph-agent",
        "input": "What's my name?",
        "previous_response_id": response_id,
        "stream": False
    }
)
# Will remember context from previous message

With Vercel AI SDK

import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'

const result = await streamText({
  model: openai('langgraph-agent'),
  baseURL: 'http://localhost:8000/v1',
  prompt: 'Hello!',  // Vercel AI SDK uses prompt
})

for await (const chunk of result.textStream) {
  console.log(chunk)
}

Advanced Configuration

from langgraph_responses_gateway import ResponsesGateway

gateway = ResponsesGateway(
    graph=your_graph,
    name="Advanced Agent",
    version="2.0.0",
    base_path="/api/v2",  # Custom base path
    model_name="my-custom-model"  # Model name reported to clients
)

# Access additional request data
class CustomGateway(ResponsesGateway):
    def _prepare_graph_input(self, user_input: str, req, previous_context=None):
        messages = []
        if previous_context:  # Continue conversation
            messages.extend(previous_context["messages"])
        messages.append({"role": "user", "content": user_input})
        
        return {
            "messages": messages,
            "thread_id": req.thread_id,  # Thread management
            "user_id": req.user_id,      # User isolation
            "metadata": req.metadata,     # Custom metadata
        }

API Endpoints

The gateway exposes the following endpoints:

POST /v1/responses

Create a response from the agent. Supports both streaming and non-streaming modes.

Request Body (OpenAI SDK Compatible):

{
  "model": "langgraph-agent",              // Required
  "input": "Your message",                // Required - string or array of input parts
  "stream": true,                         // Enable SSE streaming
  "instructions": "Be helpful",           // System instructions
  "previous_response_id": "resp_xxx",     // Chain conversations
  "store": true,                          // Store for chaining
  "temperature": 0.7,                     // Generation temperature
  "top_p": 0.9,                           // Nucleus sampling
  "max_output_tokens": 1000,              // Max tokens to generate
  "truncation": "auto",                   // Truncation strategy ("auto" or "disabled")
  "service_tier": "default",              // Processing tier
  "user": "user-123",                     // End-user identifier
  "include": ["message.output_text.logprobs"], // Additional output data
  "thread_id": "optional-thread-id",      // LangGraph thread management
  "user_id": "optional-user-id",          // User isolation (alias for 'user')
  "metadata": {}                          // Custom metadata
}

Translation Notes:

1. Message Format: The gateway automatically translates OpenAI's input parameter to LangGraph's internal messages format.

2. Conversation Management: Two different mechanisms are supported:

   • previous_response_id: OpenAI's stateless conversation continuation by referencing a specific response
   • thread_id + user: LangGraph's stateful conversation management via checkpointer (composite key = {user}:{thread_id})

   These are complementary - use previous_response_id for OpenAI-style continuation, or thread_id/user for LangGraph's persistent state management.

GET /v1/models

List available models (returns your configured model name).

GET /health

Health check endpoint for monitoring.

Customization

Custom Content Extraction

Override the content extraction methods to handle your specific LangGraph output format:

class MyGateway(ResponsesGateway):
    def _extract_content_from_step(self, step):
        # Custom logic for your graph's streaming format
        if "my_custom_field" in step:
            return step["my_custom_field"]
        return super()._extract_content_from_step(step)

Custom Input Preparation

Customize how user input is prepared for your graph:

class MyGateway(ResponsesGateway):
    def _prepare_graph_input(self, user_input: str, req, previous_context=None):
        return {
            "query": user_input,
            "history": previous_context.get("messages", []) if previous_context else [],
            "context": req.metadata.get("context", {}),
            "config": {"temperature": req.temperature or 0.7}
        }

Architecture

The gateway works by:

1. Input Translation: Converting OpenAI Responses API requests to LangGraph input format
2. Graph Execution: Running your LangGraph with the prepared input
3. Output Streaming: Converting LangGraph's streaming output to SSE events
4. Format Compliance: Ensuring all responses match OpenAI's Responses API specification

Requirements

• Python 3.9+
• LangGraph 0.5.0+
• FastAPI 0.100.0+
• Pydantic 2.0.0+

Development

Setup

# Clone the repository
git clone https://github.com/jero2rome/langgraph-responses-gateway
cd langgraph-responses-gateway

# Install with development dependencies
uv add --dev pytest pytest-asyncio pytest-cov black ruff mypy

# Run tests
uv run pytest

# Format code
uv run black src/
uv run ruff check --fix src/

Running Tests

# Run all tests
uv run pytest

# With coverage
uv run pytest --cov=langgraph_responses_gateway

# Specific test file
uv run pytest tests/test_gateway.py

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Author

Created by Jerome Mohanan

Acknowledgments

• Built on top of LangGraph by LangChain
• Implements OpenAI's Responses API specification
• Inspired by the need for better LangGraph ↔ OpenAI compatibility

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Documentation: Read the Docs (coming soon)

Roadmap

• Tool calling support (convert LangGraph tools to Responses API format)
• Reasoning output for complex agent traces
• Automatic token counting with tiktoken
• Built-in authentication and rate limiting
• Support for other agent frameworks (CrewAI, AutoGen)
• WebSocket support for bidirectional streaming
• OpenTelemetry instrumentation

Related Projects

• LangGraph - Build robust agents with LLMs
• Vercel AI SDK - Build AI-powered web applications
• FastAPI - Modern web framework for building APIs