langchain-runner 0.1.0

Run LangChain/LangGraph agents autonomously with webhooks, cron, and HTTP triggers

langchain-runner

Zero-configuration way to expose your LangChain/LangGraph agents as autonomous services.

Simply wrap your agent with Runner and it becomes a web service with webhook endpoints, cron schedules, and HTTP triggers—no infrastructure code needed.

Installation

pip install langchain-runner

Quick Start

from langchain_runner import Runner
from langchain.agents import create_agent

# Your LangChain agent (LangChain v1+)
agent = create_agent(
    model="anthropic:claude-sonnet-4-5-20250929",  # or "openai:gpt-4.1"
    tools=[],
    system_prompt="You are a helpful assistant.",
)

# Wrap it with Runner
runner = Runner(agent)

@runner.cron("0 9 * * *")
async def morning_report():
    return "Generate the daily sales report"

@runner.webhook("/stripe")
async def on_payment(payload: dict):
    return f"Process payment: {payload['type']}"

@runner.trigger("/ask")
async def ask(question: str):
    return question

runner.serve()

Features

• Webhook triggers - Receive external webhooks (GitHub, Stripe, Slack, etc.)
• Cron triggers - Schedule agent runs with cron expressions
• HTTP triggers - Simple POST endpoints to invoke your agent
• Background execution - Agent runs don't block HTTP responses
• Run tracking - Monitor agent run status and results
• Zero config - Works out of the box with sensible defaults

Endpoints

Once your runner is serving, you get these endpoints automatically:

Endpoint	Method	Description
/health	GET	Health check
/triggers	GET	List registered triggers
/trigger/{name}	POST	Invoke a registered trigger
/webhook/{name}	POST	Receive webhook payload
/runs	GET	List recent runs
/runs/{id}	GET	Get run status and result

Triggers

HTTP Trigger

Simple endpoint to invoke your agent with a custom message:

@runner.trigger("/ask")
async def ask(question: str):
    return question

curl -X POST http://localhost:8000/trigger/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the weather today?"}'

Response:

{"run_id": "abc123", "status": "pending", "message": "Run started"}

Webhook Trigger

Receive webhooks from external services (GitHub, Stripe, Clerk, etc.) and transform them into agent inputs:

@runner.webhook("/clerk")
async def on_clerk_user(payload: dict):
    event_type = payload.get("type")
    user = payload.get("data", {})
    return f"Handle Clerk event '{event_type}' for user: {user.get('email_addresses', [{}])[0].get('email_address')}"

Your webhook URL will be:

https://<your-domain>/webhook/<name>

For the example above: https://your-server.com/webhook/clerk

Setting Up Webhooks

1. For local development, expose your server with ngrok:

# Start your runner
python my_agent.py  # Starts server on port 8000

# In another terminal, expose it
ngrok http 8000
# Output: https://abc123.ngrok.io -> http://localhost:8000

Your webhook URL is now: https://abc123.ngrok.io/webhook/clerk

2. Configure the external service (example: Clerk):

1. Go to your Clerk Dashboard → Webhooks
2. Click "Add Endpoint"
3. Enter your webhook URL: https://abc123.ngrok.io/webhook/clerk
4. Select events to subscribe to: user.created, user.updated, etc.
5. Save the endpoint

3. For production, deploy your runner to a cloud provider and use that URL:

https://my-agent.railway.app/webhook/clerk

Cron Trigger

Schedule agent runs with cron expressions:

@runner.cron("0 9 * * *")  # Every day at 9am
async def daily_summary():
    return "Generate daily standup summary"

@runner.cron("0 9 * * 1-5")  # Weekdays at 9am
async def weekday_task():
    return "Process weekday reports"

@runner.cron("*/15 * * * *")  # Every 15 minutes
async def check_alerts():
    return "Check for new alerts"

Creating Your Agent

LangChain Agents (Recommended)

The recommended way to create agents is using LangChain v1+'s create_agent function:

from langchain.agents import create_agent

# Using model string (simple)
agent = create_agent(
    model="anthropic:claude-sonnet-4-5-20250929",  # or "openai:gpt-4.1"
    tools=[my_tool],
    system_prompt="You are a helpful assistant.",
)
runner = Runner(agent)

You can also use a model instance for more control:

from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    temperature=0,
    max_tokens=4096,
)
agent = create_agent(
    model=model,
    tools=[my_tool],
    system_prompt="You are a helpful assistant.",
)
runner = Runner(agent)

Custom Agents (Advanced)

For advanced use cases, you can pass any callable that accepts and returns the LangGraph message format:

# Async callable
async def my_agent(input: dict) -> dict:
    messages = input["messages"]
    # Your custom logic...
    return {"messages": [..., {"role": "assistant", "content": "response"}]}

runner = Runner(my_agent)

# Sync callable (runs in thread pool automatically)
def my_sync_agent(input: dict) -> dict:
    return {"messages": [{"role": "assistant", "content": "Hello!"}]}

runner = Runner(my_sync_agent)

Configuration

runner = Runner(
    agent,
    name="my-agent",  # Optional name (shown in /health)
    max_runs=1000,    # Max runs to keep in memory
)

runner.serve(
    host="0.0.0.0",
    port=8000,
)

Environment variables:

• LANGCHAIN_RUNNER_HOST - Host to bind to (default: 0.0.0.0)
• LANGCHAIN_RUNNER_PORT - Port to bind to (default: 8000)

CLI

# Run your agent file
langchain-runner serve my_agent.py

# Or use Python module
python -m langchain_runner serve my_agent.py

# With custom host/port
langchain-runner serve my_agent.py --host 127.0.0.1 --port 3000

Run Tracking

Every agent invocation returns a run_id immediately. Use it to track progress:

import requests

# Invoke trigger
response = requests.post(
    "http://localhost:8000/trigger/ask",
    json={"question": "Hello"}
)
run_id = response.json()["run_id"]

# Check status
status = requests.get(f"http://localhost:8000/runs/{run_id}")
print(status.json())

Response:

{
  "run_id": "abc123",
  "status": "completed",
  "trigger_type": "http",
  "trigger_name": "ask",
  "input": "Hello",
  "result": {"messages": [...]},
  "final_message": "Here's the answer...",
  "created_at": "2025-01-01T00:00:00Z",
  "completed_at": "2025-01-01T00:00:05Z"
}

Using MCP Tools

The recommended way to add tools to your agent is via Model Context Protocol (MCP). MCP provides a standard way to connect AI agents to external tools and data sources.

First, install the MCP adapter:

pip install langchain-runner[mcp]

Then connect to MCP servers and use their tools:

import asyncio
from langchain_runner import Runner
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def create_mcp_agent():
    # Connect to MCP servers
    client = MultiServerMCPClient({
        "tools": {
            "transport": "streamable_http",
            "url": "https://your-mcp-server.com/mcp",
        },
        # You can connect to multiple servers
        "local_tools": {
            "transport": "stdio",
            "command": "python",
            "args": ["./my_mcp_server.py"],
        },
    })
    
    # Get tools from all connected servers
    tools = await client.get_tools()
    print(f"Loaded {len(tools)} tools from MCP servers")
    
    # Create agent with MCP tools
    agent = create_agent(
        model="anthropic:claude-sonnet-4-5-20250929",
        tools=tools,
        system_prompt="You are a helpful assistant with access to external tools.",
    )
    
    return agent

# Create agent at startup
agent = asyncio.get_event_loop().run_until_complete(create_mcp_agent())
runner = Runner(agent, name="mcp-agent")

@runner.webhook("/process")
async def on_process(payload: dict):
    return f"Process this request: {payload}"

if __name__ == "__main__":
    runner.serve()

Example: Full Setup

import asyncio
from langchain_runner import Runner
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def setup_agent():
    # Connect to your MCP server with tools (Notion, Slack, etc.)
    client = MultiServerMCPClient({
        "tools": {
            "transport": "streamable_http",
            "url": "https://your-mcp-server.com/mcp",
        }
    })
    tools = await client.get_tools()
    
    return create_agent(
        model="anthropic:claude-sonnet-4-5-20250929",
        tools=tools,
        system_prompt="You are a helpful assistant.",
    )

agent = asyncio.get_event_loop().run_until_complete(setup_agent())
runner = Runner(agent, name="my-assistant")

# HTTP trigger
@runner.trigger("/chat")
async def chat(message: str):
    return message

# Webhook triggers
@runner.webhook("/slack")
async def on_slack(payload: dict):
    event = payload.get("event", {})
    return f"Respond to: {event.get('text', '')}"

@runner.webhook("/github")
async def on_github(payload: dict):
    pr = payload.get("pull_request", {})
    return f"Review PR: {pr.get('title', 'Unknown')}"

# Cron triggers
@runner.cron("0 9 * * 1-5")  # Weekdays at 9am
async def daily_standup():
    return "Generate daily standup summary"

@runner.cron("0 18 * * 5")  # Friday at 6pm
async def weekly_report():
    return "Generate weekly report"

if __name__ == "__main__":
    runner.serve()

License

MIT