Build production-grade AI applications with stable outputs, composable agents, observable actions, and testable workflows.

English | 中文介绍

🔥 Docs · 🚀 Quickstart · 🏗️ Architecture · 💡 Capabilities · 🧩 Ecosystem

LangChain, CrewAI, and AutoGen each solve a real problem — but they optimize for exploration, not delivery. Teams that ship AI-powered products into production consistently run into the same walls:

Agently is designed from the start for the gap between "works in a notebook" and "runs reliably in production":

• Stable outputs — contract-first schema with mandatory field enforcement and automatic retries
• Testable orchestration — every TriggerFlow branch is a plain Python function, independently unit-testable
• Observable actions — every tool/MCP/sandbox call is logged with input, output, and timing
• Pause, resume, persist — TriggerFlow executions can be saved to disk and restored after process restart
• Project-scale config — hierarchical YAML/TOML/JSON settings files, env-variable substitution, and scaffolding via agently-devtools init

Agently 4.1 adds a fully rewritten Action Runtime: a three-layer extensible plugin stack (planning → loop → execution) with native support for local functions, MCP servers, Python/Bash sandboxes, and custom backends.

Agently organizes every AI application into four clear layers. Each layer has a stable interface — replaceable, extendable, and independently testable.

graph TB
    subgraph APP["Your Application"]
        UserCode["Application / Business Logic"]
    end

    subgraph GLOBAL["Agently Global"]
        Global["Global Settings  ·  Plugin Manager  ·  Default Model Config"]
    end

    subgraph AGENT["Agent Layer"]
        direction TB
        AgentInst["Agent Instance"]
        subgraph AGENT_INNER["Per-Agent Components"]
            Prompt["Prompt\ninput · instruct · info · output"]
            Session["Session\nmulti-turn memory · memo · persistence"]
            Action["Action\nplanning · dispatch · logs"]
            Settings2["Hierarchical Settings\n(inherits from global)"]
        end
    end

    subgraph MODEL["Model Layer"]
        ModelReq["Model Request\nbuilt from prompt slots + settings"]
        ModelResp["Model Response\nstructured · streaming · instant events"]
    end

    subgraph WORKFLOW["Workflow Orchestration"]
        TF["TriggerFlow\nto · if/elif/else · match/case · batch · for_each · when · emit · pause/resume · persist"]
    end

    subgraph LLMS["LLM APIs"]
        LLMAPIs["OpenAI · DeepSeek · Claude · Qwen · Ollama · any OpenAI-compatible endpoint"]
    end

    UserCode -->|"configure & create"| Global
    Global -->|"inherited settings"| AgentInst
    AgentInst --- AGENT_INNER
    AgentInst -->|"build & send"| ModelReq
    ModelReq -->|"HTTP"| LLMAPIs
    LLMAPIs -->|"response stream"| ModelResp
    UserCode -->|"orchestrate"| TF
    TF -->|"trigger agent steps"| AgentInst

Three independently replaceable layers — swap only what you need.

graph LR
    subgraph AGENT2["Agent"]
        AExt["ActionExtension\nprepares visible actions & logs"]
    end

    subgraph RUNTIME["Action Runtime (replaceable)"]
        AR["ActionRuntime Plugin\nAgentlyActionRuntime\n\nplanning protocol · call normalization · round control"]
        AF["ActionFlow Plugin\nTriggerFlowActionFlow\n\naction loop · pause/resume · concurrency"]
    end

    subgraph EXECUTORS["ActionExecutor (replaceable)"]
        E1["LocalFunctionExecutor\n@action_func / @tool_func"]
        E2["MCPExecutor\nstdio · http"]
        E3["PythonSandboxExecutor"]
        E4["BashSandboxExecutor"]
        E5["Custom Plugin"]
    end

    AExt -->|"delegate planning"| AR
    AR -->|"run action loop"| AF
    AF -->|"dispatch call"| E1
    AF -->|"dispatch call"| E2
    AF -->|"dispatch call"| E3
    AF -->|"dispatch call"| E4
    AF -->|"dispatch call"| E5

Define the schema once. Agently enforces it on every call, with automatic retries when critical fields are missing.

result = (
    agent
    .input("Analyze this review: 'Great product, but slow shipping.'")
    .output({
        "sentiment": (str, "positive / neutral / negative"),
        "key_issues": [(str, "issue summary")],
        "priority": (int, "1–5, 5 is most urgent"),
    })
    .start(ensure_keys=["sentiment", "key_issues[*]"])
)
# Always a dict — "sentiment" and every "key_issues" item guaranteed present

Each output field streams independently. Drive UI updates or downstream logic as fields complete, not after the whole response.

response = (
    agent
    .input("Explain recursion with 3 examples")
    .output({
        "definition": (str, "one-sentence definition"),
        "examples": [(str, "code example with explanation")],
    })
    .get_response()
)

for event in response.get_generator(type="instant"):
    if event.path == "definition" and event.delta:
        ui.update_header(event.delta)             # stream definition character by character
    if event.wildcard_path == "examples[*]" and event.is_complete:
        ui.append_example(event.value)            # append each complete example

Mount any combination. The runtime handles planning, execution, retries, and full structured logs.

@agent.action_func
def search_docs(query: str) -> str:
    """Search internal documentation."""
    return docs_db.search(query)

agent.use_mcp("docs-server", transport="stdio", command=["python", "mcp_server.py"])
agent.use_sandbox("python")                          # isolated Python execution

agent.use_actions([search_docs, "docs-server", "python"])

response = agent.input("Find auth docs and show a login code example.").get_response()

# Every call: what was invoked, with what args, what it returned
print(response.result.full_result_data["extra"]["action_logs"])

Legacy tool APIs (@agent.tool_func, agent.use_tool()) continue to work and map to the same runtime.

TriggerFlow goes well beyond chaining functions. It's a full workflow engine with concurrency, event-driven branching, human-in-the-loop interrupts, and execution persistence.

Concurrency — batch and for_each

Run steps in parallel with a configurable concurrency limit:

# Process a list of URLs, max 5 in parallel
(
    flow.for_each(url_list, concurrency=5)
    .to(fetch_page)
    .to(summarize)
    .end()
)

# Fan out to N fixed branches simultaneously
flow.batch(3).to(strategy_a).collect("results", "a")
flow.batch(3).to(strategy_b).collect("results", "b")
flow.batch(3).to(strategy_c).collect("results", "c")

Event-driven — when and emit

Branch on signals, chunk completions, or data changes — not just linear sequence:

flow.when("UserInput").to(process_input).to(plan_next_step)
flow.when("ToolResult").to(evaluate_result)
flow.when({"runtime_data": "user_decision"}).to(apply_decision)

# Emit from inside a chunk to trigger other branches
async def plan_next_step(data: TriggerFlowEventData):
    if needs_tool:
        await data.async_emit("ToolCall", tool_args)
    else:
        await data.async_emit("UserInput", final_reply)

Pause, Resume, and Persistence

Save execution state to disk, restore it after a process restart — critical for long-running or human-in-the-loop workflows:

# Start execution, save checkpoint immediately
execution = flow.start_execution(initial_input, wait_for_result=False)
execution.save("checkpoint.json")

# Later — new process, restored state, continue from where it paused
restored = flow.create_execution()
restored.load("checkpoint.json")
restored.emit("UserFeedback", {"approved": True, "note": "Looks good."})
result = restored.get_result(timeout=30)

This makes Agently workflows genuinely restart-safe — suitable for approval gates, multi-day pipelines, and human review loops.

Blueprint Serialization

The flow topology itself can be exported to JSON/YAML and reloaded, enabling dynamic workflow definitions and versioned flow configurations:

flow.get_yaml_flow(save_to="flows/main_flow.yaml")
# later
flow.load_flow_config("flows/main_flow.yaml")

Activate a session by ID. Agently maintains chat history, applies window trimming, supports custom memo strategies, and persists state to JSON/YAML.

agent.activate_session(session_id="user-42")
agent.set_settings("session.max_length", 10000)

session = agent.activated_session
session.register_analysis_handler(decide_when_to_summarize)
session.register_resize_handler("summarize_oldest", summarize_handler)

reply1 = agent.input("My name is Alice.").start()
reply2 = agent.input("What's my name?").start()   # correctly returns "Alice"

Real AI projects involve multiple agents, multiple prompt templates, and multiple environments. Agently's hierarchical settings system supports YAML/JSON/TOML config files at every layer, with ${ENV.VAR} substitution and layered inheritance.

Recommended project structure:

my_ai_project/
├── .env                          # API keys and secrets
├── config/
│   ├── global.yaml               # global model + runtime settings
│   └── agents/
│       ├── researcher.yaml       # per-agent model overrides
│       └── writer.yaml
├── prompts/
│   ├── researcher_role.yaml      # reusable prompt templates
│   └── writer_role.yaml
├── flows/
│   ├── main_flow.py              # TriggerFlow definitions
│   └── main_flow.yaml            # serialized flow blueprint (optional)
├── agents/
│   ├── researcher.py
│   └── writer.py
└── main.py

Config hierarchy — each layer inherits and overrides:

graph LR
    A["global.yaml\n(model, runtime defaults)"]
    B["agents/researcher.yaml\n(per-agent model overrides)"]
    C["agent.set_settings(...)\n(per-request overrides)"]

    A -->|"inherited by"| B
    B -->|"inherited by"| C

# global.yaml loaded once at startup
Agently.load_settings("yaml", "config/global.yaml", auto_load_env=True)

# each agent loads its own overrides
researcher = Agently.create_agent()
researcher.load_settings("yaml", "config/agents/researcher.yaml")

# request-level override if needed
researcher.set_settings("OpenAICompatible.request_options", {"temperature": 0.2})

Scaffold a new project instantly:

pip install agently-devtools
agently-devtools init my_project

See real-world project examples:

• Agently-Daily-News-Collector — scheduled multi-source news pipeline
• Agently-Talk-to-Control — conversational control flow with TriggerFlow

Prompts are structured slots, not raw strings. Each slot has a role: input (the task), instruct (constraints), info (context data), output (schema). Agent-level slots persist across requests; request-level slots apply once.

agent.role("You are a senior Python code reviewer.")   # always present

result = (
    agent
    .input(user_code)
    .instruct("Focus on security and performance.")
    .info({"context": "Public-facing API handler", "framework": "FastAPI"})
    .output({"issues": [(str, "issue description")], "score": (int, "0–100")})
    .start()
)

Prompt templates can be loaded from YAML/JSON files via the configure_prompt extension for team-level prompt governance.

One config object, any provider, no vendor lock-in.

Agently.set_settings(
    "OpenAICompatible",
    {
        "base_url": "https://api.deepseek.com/v1",
        "model": "deepseek-chat",
        "auth": "DEEPSEEK_API_KEY",   # reads from env automatically
    },
)
# Change base_url + model to switch providers — business code unchanged

Supported: OpenAI · DeepSeek · Anthropic Claude (via proxy) · Qwen · Mistral · Llama · local Ollama · any OpenAI-compatible endpoint.

pip install -U agently

Python ≥ 3.10 required.

from agently import Agently

Agently.set_settings("OpenAICompatible", {
    "base_url": "https://api.deepseek.com/v1",
    "model": "deepseek-chat",
    "auth": "DEEPSEEK_API_KEY",
})

agent = Agently.create_agent()

result = (
    agent.input("Introduce Python in one sentence and list 3 strengths")
    .output({
        "intro": (str, "one sentence"),
        "strengths": [(str, "strength")],
    })
    .start(ensure_keys=["intro", "strengths[*]"])
)

print(result)
# {"intro": "Python is ...", "strengths": ["...", "...", "..."]}

Official Agently Skills give AI coding assistants (Claude Code, Cursor, etc.) the knowledge to implement Agently patterns correctly, without re-explaining the framework each session.

• Repo: https://github.com/AgentEra/Agently-Skills
• Install: npx skills add AgentEra/Agently-Skills

Covers: single-request design · TriggerFlow orchestration · multi-agent · MCP · Session · FastAPI integration · LangChain/LangGraph migration playbooks.

agently-devtools is an optional companion package for runtime inspection and project scaffolding.

pip install agently-devtools
agently-devtools init my_project    # scaffold a new Agently project

• Runtime observation: ObservationBridge, create_local_observation_app
• Examples: examples/devtools/
• Compatibility: agently-devtools 0.1.x targets agently >=4.1.0,<4.2.0

Agently is designed to be extended at multiple independent levels. You don't have to fork the framework to change how it behaves — every major component is a replaceable plugin, hook, or registered handler.

graph TB
    subgraph USER["Your Application"]
        App["Business Logic  ·  TriggerFlow Orchestration"]
    end

    subgraph AGENT_EXT["Agent Extension Layer"]
        AE["Built-in Extensions\nSession · Action · AutoFunc · ConfigurePrompt · KeyWaiter · StreamingPrint"]
        AEC["Custom Agent Extension\nregister your own lifecycle hooks\nand agent-level capabilities"]
    end

    subgraph HOOKS["Request Lifecycle Hooks"]
        H1["request_prefixes\nmodify Prompt + Settings before request"]
        H2["broadcast_prefixes\nreact when response starts"]
        H3["broadcast_suffixes\nreact per streaming event"]
        H4["finally\nrun after response completes"]
    end

    subgraph CORE_PIPELINE["Core Pipeline Plugins (replaceable)"]
        PG["PromptGenerator\ncontrols how slots are assembled into the final prompt"]
        MR["ModelRequester\nOpenAICompatible (default) · custom provider"]
        RP["ResponseParser\ncontrols how raw model output is parsed"]
    end

    subgraph ACTION_STACK["Action Runtime Plugins (replaceable)"]
        AR["ActionRuntime\nplanning protocol · round control"]
        AF["ActionFlow\nloop shape · pause/resume · concurrency"]
        AX["ActionExecutor\nLocalFunction · MCP · PythonSandbox · BashSandbox · Custom"]
    end

    subgraph TF_EXT["TriggerFlow Extensions"]
        TC["Custom Chunk Handlers\n@flow.chunk or register_chunk_handler"]
        TCC["Custom Condition Handlers\nregister_condition_handler"]
    end

    subgraph HOOKERS["Runtime Event Hookers"]
        RH["ConsoleSink · StorageSink · ChannelSink · Custom\nattach to any runtime event stream"]
    end

    App --> AGENT_EXT
    AGENT_EXT --> HOOKS
    HOOKS --> CORE_PIPELINE
    CORE_PIPELINE --> ACTION_STACK
    App --> TF_EXT
    CORE_PIPELINE -.->|"emit runtime events"| HOOKERS
    ACTION_STACK -.->|"emit runtime events"| HOOKERS

from agently.types.plugins import ActionExecutor, ActionRunContext, ActionExecutionRequest, ActionResult

class MyCloudExecutor:
    name = "my-cloud-executor"
    DEFAULT_SETTINGS = {}

    async def execute(
        self,
        context: ActionRunContext,
        request: ActionExecutionRequest,
    ) -> list[ActionResult]:
        # call your cloud function / RPC / custom backend
        ...

Agently.plugin_manager.register("ActionExecutor", MyCloudExecutor)

agent = Agently.create_agent()

# Inject context into every request this agent makes
def inject_tenant_context(prompt, settings):
    prompt.info({"tenant_id": get_current_tenant()})

agent.extension_handlers.append("request_prefixes", inject_tenant_context)

The term AI application harness describes a layer that wraps LLM calls with engineering controls — stable interfaces, observable internals, pluggable components. It's an architectural quality, not a product category.

Agently is a development framework, but it's designed to satisfy exactly those properties:

These properties are a consequence of Agently's design philosophy — what you get when you structure an AI application the Agently way.

"Agently helped us turn evaluation rules into executable workflows and keep key scoring accuracy at 75%+, significantly improving bid-evaluation efficiency." — Project lead at a large energy SOE

"Agently enabled a closed loop from clarification to query planning to rendering, reaching 90%+ first-response accuracy and stable production performance." — Data lead at a large energy group

"Agently's orchestration and session capabilities let us ship a teaching assistant for course management and Q&A quickly, with continuous iteration." — Project lead at a university teaching-assistant initiative

📢 Share your case on GitHub Discussions →

Q: What makes Agently different from LangChain? LangChain is excellent for prototyping and has a broad ecosystem. Agently is optimized for the post-prototype phase: contract-first outputs prevent interface drift, TriggerFlow branches are individually unit-testable, and the project-scale config system supports real engineering workflows. If you've shipped with LangChain and hit maintainability walls, Agently is built for exactly that.

Q: How is Agently different from CrewAI or AutoGen? CrewAI and AutoGen are designed around agent teams with natural-language coordination — great for exploration, hard to make deterministic. Agently uses explicit code-based orchestration (TriggerFlow) where every branch is a Python function with clear inputs and outputs, every action call is logged, and executions can be paused, serialized, and resumed — properties that matter when you're shipping to users.

Q: What is the Action Runtime, and why was it rewritten in v4.1? The old tool system was a single flat layer — enough for simple use cases, but not extensible. The new Action Runtime separates planning ("what to call"), loop orchestration ("how many rounds, with what concurrency"), and execution ("actually run the function/MCP/sandbox"). Each layer is a plugin. You can swap just the sandbox backend without touching the planning logic, or replace just the planning algorithm without changing how loops run.

Q: How do I deploy an Agently service? Agently doesn't prescribe a deployment model. It provides full async APIs. The examples/fastapi/ directory covers SSE streaming, WebSocket, and standard POST. See Agently-Talk-to-Control for a complete deployed example.

Q: Is there enterprise support? Yes. The core framework is open-source under Apache 2.0. Enterprise extensions, private deployment support, governance modules, and SLA-backed collaboration are available under separate commercial agreements. Contact us via the community.

• Discussions: https://github.com/AgentEra/Agently/discussions
• Issues: https://github.com/AgentEra/Agently/issues
• WeChat Group: https://doc.weixin.qq.com/forms/AIoA8gcHAFMAScAhgZQABIlW6tV3l7QQf