LangChain and LlamaIndex
Turn Anlyon actions into LangChain or LlamaIndex tools with the anlyon-langchain and anlyon-llamaindex packages, bind the tools to one of your users, and trace LangChain invocations as Anlyon runs.
Two PyPI packages put your Anlyon actions into a Python agent framework. Each builds one tool per enabled action, with the action’s own name, description and input schema. When the model calls a tool, the tool calls client.actions.invoke(): Anlyon adds the credential, applies your policies and limits, makes the call and returns the invocation with its receipt grade. The tool never holds a token.
| Package | Version | Requires | Covered here |
|---|---|---|---|
anlyon-langchain |
0.4.0 | anlyon>=0.8.0, langchain-core>=1.6.6 |
action_tools, AnlyonTracer |
anlyon-llamaindex |
0.4.0 | anlyon>=0.8.0, llama-index-core>=0.14.25 |
action_tools |
Both need Python 3.10 or later, and both build on the Python SDK.
Before you start
- Declare the actions the agent may call. See Define an action, or run the quickstart, which declares the
open-issueaction used below. - The key that builds the tools needs
actions:read, to list the actions, andactions:invoke, to call them.AnlyonTraceralso needsruns:write. - To act for your users, use an app key with
end-users:write, andend-users:approvalsif your app shows the user their approvals. See Act for one of your users.
Both packages take the synchronous Client, not AsyncClient.
LangChain
pip install anlyon-langchain
The example below also uses langchain and langchain-openai.
Give an agent your actions
action_tools(client, names=None) returns a list of LangChain StructuredTools. Build them from client.for_user(user_id), with the user from your own session, and every call uses that user’s connections, limits and approvals.
import os
from anlyon import Client
from anlyon_langchain import action_tools
from langchain.agents import create_agent
anlyon = Client(api_key=os.environ["ANLYON_APP_KEY"])
def build_agent(session_user_id: str):
# Your server names the user. No tool has a user field the model could change.
user = anlyon.for_user(session_user_id)
return create_agent(
model="openai:gpt-5-mini",
tools=action_tools(user, names=["open-issue"]),
system_prompt="You help the user track bugs. Call open-issue once per issue.",
)
agent = build_agent("alice")
result = agent.invoke(
{"messages": [{"role": "user", "content": "Open an issue: checkout fails on Safari."}]}
)
print(result["messages"][-1].content)
To bind the tools to a chat model yourself, pass them to model.bind_tools(tools). To pause a LangGraph graph while a call waits for approval, see LangGraph.
Call a tool directly
A tool returns the invocation as a dict:
(open_issue,) = action_tools(anlyon.for_user("alice"), names=["open-issue"])
result = open_issue.invoke({"title": "Checkout fails on Safari", "body": "Steps to reproduce: ..."})
if result.get("pendingApproval"):
print("Waiting for a decision:", result["approvalId"])
elif result["status"] in ("connect_required", "reconnect_required"):
print("Needs a connection:", result["connectUrl"])
else:
print(result["status"], result["grade"]) # e.g. succeeded confirmed
Trace an invocation as a run
AnlyonTracer is a LangChain callback handler. Pass it in a run’s callbacks, and the invocation is recorded as an Anlyon run with nested spans.
from anlyon_langchain import AnlyonTracer
tracer = AnlyonTracer(anlyon, application="support-agent")
result = agent.invoke(
{"messages": [{"role": "user", "content": "Where is my order?"}]},
config={"callbacks": [tracer]},
)
| Argument | Type | Default | Meaning |
|---|---|---|---|
client |
Client |
required | The client that writes the run. Its key needs runs:write. |
application |
str |
"langchain" |
The run’s application. |
run_name |
str |
"LangChain run" |
The run’s name. |
The root callback starts one run, with the root’s tags and its metadata plus {"framework": "langchain"}. Each nested callback becomes a span linked to its parent:
| LangChain callback | Span type | Recorded |
|---|---|---|
| Chain | workflow_step |
Inputs and outputs. |
| LLM or chat model | model_call |
Prompts, output, token usage and model name when the provider reports them. |
| Tool | tool_call |
Input and output. |
A span ends succeeded or failed, with the error message on failure. When the root finishes, the run ends with the same status and the root’s output. The tracer is best-effort: if Anlyon cannot be reached, the LangChain result does not change. One tracer can be shared across concurrent invocations.
The tools from action_tools do not pass the tracer’s run id to actions.invoke, so the invocation and its receipt are not joined to the run’s timeline. To join them, invoke the action yourself with run_id, as in Trace a run.
LlamaIndex
pip install anlyon-llamaindex
The example below also uses llama-index-llms-openai.
Give an agent your actions
action_tools(client, names=None) returns a list of LlamaIndex FunctionTools. Each tool’s parameters are the action’s input schema as it is declared.
import asyncio
import os
from anlyon import Client
from anlyon_llamaindex import action_tools
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
anlyon = Client(api_key=os.environ["ANLYON_APP_KEY"])
async def chat(session_user_id: str, text: str) -> str:
# Your server names the user. No tool has a user field the model could change.
tools = action_tools(anlyon.for_user(session_user_id), names=["open-issue"])
agent = FunctionAgent(
tools=tools,
llm=OpenAI(model="gpt-5-mini"),
system_prompt="You help the user track bugs. Call open-issue once per issue.",
)
response = await agent.run(user_msg=text)
return str(response)
print(asyncio.run(chat("alice", "Open an issue: checkout fails on Safari.")))
Call a tool directly
The tool’s raw_output is the invocation as a dict:
(open_issue,) = action_tools(anlyon.for_user("alice"), names=["open-issue"])
output = open_issue.call(title="Checkout fails on Safari", body="Steps to reproduce: ...")
result = output.raw_output
print(open_issue.metadata.name, result["status"], result["grade"])
How the tools behave
The two packages share one implementation in the anlyon SDK, so a tool reads and behaves the same in either framework.
| What the tool does | |
|---|---|
| Name | The action’s name, such as open-issue. |
| Description | The action’s description. When the action declares an impact, a sentence follows, such as Declared impact: emails. Each call counts toward any shared limit set for this environment. It says what one call counts, not whether a limit is set. An action with no description reads Invoke the Anlyon action <name>. |
| Input | The action’s input schema. An action without one takes no input. No schema has an end-user field. |
| Which actions | Every enabled action, or the ones in names. Disabled actions are left out. |
| Retries | The SDK sends one idempotency key per tool call and reuses it on its own transport retries, so a resend is not run twice. Two tool calls are two invocations. |
Read the result
| The result has | What it means | What to do |
|---|---|---|
status and grade |
The call ran. grade is the receipt grade, such as confirmed, acknowledged, unknown, failed, refused or denied. It is None for an action with no declaration. |
Report it. See Receipts and grades. |
pendingApproval: True and approvalId, grade pending |
A policy parked the call for a decision. Nothing was sent. | Show the approval to whoever decides it: the user, or an approver on your team, as the policy says. Do not call the tool again. See Approvals. |
status: "connect_required" |
The user has not connected the integration. Nothing was sent. | Render connectUrl as a Connect button. |
status: "reconnect_required" |
The provider revoked the user’s connection. Nothing was sent. | The same, to reconnect. |
A connect result looks like this:
{
"status": "connect_required",
"integration": "github",
"connectUrl": "https://...",
"message": "The user has not connected github yet, so nothing was done. Ask them to connect it, then try again."
}
connectUrl is None when the environment has no default redirect URL. Then create a connect link yourself and confirm the connection in your callback, as in Connect a user’s account.
Any other error raises the SDK’s exception, as a direct call would. For example, a tool built from a workspace client, with a key set to require an end user, raises EndUserRequiredError (400 END_USER_REQUIRED). See Handle errors.
Related
- Python SDK: the client both packages use.
- LangGraph: pause a graph while the user approves, then resume it.
- Workspace connections and end-user connections.
- Receipts and grades and Idempotency and retries.

