langchain-agent-server 0.1.0
Lightweight server for LangChain and LangGraph agents. Serve any agent as a REST API with invoke and streaming endpoints.
Requires Python: >=3.11
Keywords: agent, fastapi, langchain, langgraph, server, sse

# langchain-agent-server

Lightweight server for [LangChain](https://www.langchain.com/) and [LangGraph](https://www.langchain.com/langgraph) agents. Serve any agent as a REST API with invoke and streaming endpoints.

No vendor lock-in. No paid platform. Just FastAPI + SSE.

## Install

```bash
pip install langchain-agent-server
```

## Quick Start

```python
from langchain_agent_server import create_app, NoAuth
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# Build your agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools=[...])

# Wrap it in a server
async def agent_factory(auth_context):
    return agent

app = create_app(
    title="My Agent API",
    agent_factory=agent_factory,
    auth=NoAuth(),
)
```

Run with any ASGI server:

```bash
uvicorn myapp:app --host 0.0.0.0 --port 8000
```

Or directly:

```python
if __name__ == "__main__":
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", "8124"))

    uvicorn.run(app, host=host, port=port, log_level="info")
```

## Endpoints

### `GET /health`

Health check. Returns `{"status": "ok"}`.

### `POST /invoke`

Invoke the agent and get all response messages at once.

**Request:**

```json
{
  "input": {
    "messages": [
      {"role": "user", "content": "What's the weather in NYC?"}
    ]
  }
}
```

**Response:**

```json
{
  "result": {
    "messages": [
      {"type": "human", "role": "user", "content": "What's the weather in NYC?"},
      {"type": "ai", "role": "assistant", "content": "", "tool_calls": [
        {"name": "get_weather", "args": {"city": "NYC"}}
      ]},
      {"type": "tool", "role": "tool", "name": "get_weather", "content": "72°F, sunny"},
      {"type": "ai", "role": "assistant", "content": "It's 72°F and sunny in NYC!"}
    ]
  }
}
```

### `POST /stream`

Invoke the agent and stream events via Server-Sent Events (SSE).

Same request format as `/invoke`. Returns `text/event-stream`.

**Events:**

```
data: {"type": "token", "content": "It's"}

data: {"type": "token", "content": " 72"}

data: {"type": "tool_start", "name": "get_weather", "input": {"city": "NYC"}}

data: {"type": "tool_end", "name": "get_weather", "output": "72°F, sunny"}

data: {"type": "token", "content": "It's 72°F and sunny!"}

data: {"type": "end", "messages": [...]}

```

Event types:

| Type | Description |
|------|-------------|
| `token` | A text chunk streamed from the LLM |
| `tool_start` | Agent is calling a tool (includes `name` and `input`) |
| `tool_end` | Tool returned a result (includes `name` and `output`) |
| `error` | An error occurred (includes `error` and `error_type`) |
| `end` | Stream complete (includes final `messages` array) |

## Authentication

### Bearer Token (default)

```python
from langchain_agent_server import create_app, BearerAuth

async def agent_factory(token: str):
    # token is the Bearer token from the Authorization header
    # Use it to create a per-request agent with user context
    return build_agent_for_user(token)

app = create_app(
    title="My Agent API",
    agent_factory=agent_factory,
    auth=BearerAuth(),  # this is the default
)
```

Returns 403 if no `Authorization` header is provided.

### No Auth

```python
from langchain_agent_server import create_app, NoAuth

async def agent_factory(auth_context):
    # auth_context is always None
    return my_agent

app = create_app(
    title="My Agent API",
    agent_factory=agent_factory,
    auth=NoAuth(),
)
```

### Custom Auth

Implement the `AuthDependency` protocol:

```python
from fastapi import HTTPException

class ApiKeyAuth:
    def __init__(self, valid_keys: set[str]):
        self.valid_keys = valid_keys

    async def __call__(self, authorization: str | None) -> str:
        if authorization not in self.valid_keys:
            raise HTTPException(status_code=401, detail="Invalid API key")
        return authorization

app = create_app(
    title="My Agent API",
    agent_factory=agent_factory,
    auth=ApiKeyAuth({"sk-secret-key-1", "sk-secret-key-2"}),
)
```

The auth callable receives the raw `Authorization` header and returns whatever your `agent_factory` needs.

## Message Conversion

The package includes utilities for converting between OpenAI and LangChain message formats:

```python
from langchain_agent_server import (
    convert_openai_to_langchain_messages,
    convert_langchain_to_openai_message,
)

# OpenAI format -> LangChain objects
messages = convert_openai_to_langchain_messages([
    {"role": "user", "content": "Hello"},
    {"role": "assistant", "content": "Hi!"},
])

# LangChain object -> OpenAI format dict
from langchain_core.messages import AIMessage
msg_dict = convert_langchain_to_openai_message(AIMessage(content="Hi!"))
# {"type": "ai", "role": "assistant", "content": "Hi!"}
```

## Why?

LangChain's official options for serving agents are:

- **LangServe** — maintenance mode, no new features
- **LangGraph Platform** — paid managed service (self-hosted requires Enterprise license)

This package is a lightweight alternative: ~200 lines of code, zero vendor lock-in, MIT licensed. You own your infrastructure.

## License

MIT