langchain-rabbitmq

Production-grade LangChain tools for RabbitMQ.
Give your AI agents the full power of RabbitMQ — queue management, exchange routing, message operations, and broker monitoring — through 21 structured LangChain tools.

Table of contents

• Features
• Installation
• Quick start
• Tools reference
• Configuration
• Advanced usage
• Compatibility
• Contributing
• License

Features

• 21 LangChain tools covering every RabbitMQ operation an agent needs
• Pydantic v2 schemas — every argument is validated before touching the broker
• Sync and async — _run / _arun on every tool; backed by pika and aio-pika
• LCEL / Runnable compatible — drop tools straight into any LangChain chain
• Structured error types — RabbitMQConnectionError, RabbitMQChannelError, RabbitMQMessageError, RabbitMQAdminError surface as agent-readable strings, never raw tracebacks
• Retry logic via tenacity with configurable back-off
• SSL/TLS support for encrypted broker connections
• Management API integration for monitoring and admin tasks
• ≥ 90 % test coverage — unit, integration (Testcontainers), and E2E agent tests

Installation

pip install langchain-rabbitmq

For development (tests, linting, docs):

pip install "langchain-rabbitmq[dev,docs]"
# or
make install

Quick start

import os
from langchain_rabbitmq import RabbitMQToolkit

# Credentials read from environment variables automatically
os.environ["RABBITMQ_URL"] = "amqp://guest:guest@localhost:5672/"

toolkit = RabbitMQToolkit.from_settings()
tools   = toolkit.get_tools()          # returns all 21 BaseTool instances

# Use with any LangChain agent
from langchain.agents import create_react_agent
agent = create_react_agent(llm=your_llm, tools=tools, prompt=your_prompt)

Minimal environment

docker run -d --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:3.12-management

Tools reference

Configuration

All settings are loaded from environment variables via RabbitMQSettings (backed by pydantic-settings).

Programmatic configuration

from langchain_rabbitmq.config import RabbitMQSettings
from langchain_rabbitmq import RabbitMQToolkit

settings = RabbitMQSettings(
    host="broker.internal",
    port=5672,
    virtual_host="production",
    username="app",
    password="s3cr3t",
    ssl=True,
    management_api_url="http://broker.internal:15672",
)
toolkit = RabbitMQToolkit(settings=settings)

Advanced usage

Async agent

from langchain_rabbitmq import RabbitMQToolkit

toolkit = RabbitMQToolkit.from_settings()
tools   = toolkit.get_tools()

# All tools implement _arun — use with async LangChain runtimes
result = await tools[0].ainvoke({"name": "orders", "durable": True})

Individual tools

from langchain_rabbitmq.tools.queue import DeclareQueueTool
from langchain_rabbitmq.config import RabbitMQSettings

tool   = DeclareQueueTool(settings=RabbitMQSettings())
result = tool.invoke({"name": "orders", "durable": True})
print(result)  # "Queue 'orders' declared successfully."

Error handling

All errors derive from RabbitMQToolException and are returned to the agent as human-readable strings. The agent can react and retry without seeing raw stack traces.

from langchain_rabbitmq.exceptions import (
    RabbitMQConnectionError,
    RabbitMQChannelError,
    RabbitMQMessageError,
    RabbitMQAdminError,
)

Compatibility

Contributing

See CONTRIBUTING.md for commit conventions, branch strategy, and how to run the test suite.

make test          # all tests (requires Docker for integration tests)
make test-unit     # unit tests only — no broker needed
make lint          # ruff check
make typecheck     # pyrefly strict
make security      # bandit scan

License

MIT © 2024 MiltonJ23