AgentFlow is a production-grade Python framework where autonomous AI agents dynamically orchestrate, compose, and self-heal API workflows across enterprise integration platforms — with first-class MuleSoft Anypoint support.
Modern enterprises run hundreds of APIs across MuleSoft, AWS API Gateway, Azure APIM, and custom services. Composing these APIs into reliable workflows requires:
- Static orchestration that breaks when APIs change
- Manual error handling per integration point
- No intelligent routing based on latency, cost, or capability
- Zero natural-language accessibility for non-technical stakeholders
AgentFlow introduces autonomous AI agents that understand API capabilities semantically and can:
- Parse natural-language intents into executable API workflows
- Dynamically discover and compose APIs at runtime
- Route intelligently based on latency, cost, rate limits, and capability matching
- Self-heal with circuit breakers, adaptive retries, and fallback chains
- Collaborate via a multi-agent protocol for complex cross-platform orchestrations
┌─────────────────────────────────────────────────┐
│ Intent Layer │
│ Natural Language → Structured API Plan │
├─────────────────────────────────────────────────┤
│ Agent Orchestrator │
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ Planner │ │ Executor │ │ Validator │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └──────────┘ └──────────┘ └──────────────┘ │
├─────────────────────────────────────────────────┤
│ Dynamic Router │
│ Latency │ Cost │ Rate Limit │ Capability │
├─────────────────────────────────────────────────┤
│ Resilience Layer │
│ Circuit Breaker │ Retry │ Fallback │ Bulkhead │
├─────────────────────────────────────────────────┤
│ Connector Layer │
│ MuleSoft │ REST │ GraphQL │ gRPC │ Custom │
└─────────────────────────────────────────────────┘
from agentflow import AgentOrchestrator, MuleSoftConnector
# Initialize with MuleSoft Anypoint
orchestrator = AgentOrchestrator(
connectors=[
MuleSoftConnector(
anypoint_url="https://anypoint.mulesoft.com",
org_id="your-org-id",
environment="production"
)
]
)
# Natural language orchestration
result = await orchestrator.execute(
"Fetch customer 12345 from CRM, enrich with credit score, "
"and create a loan application if score > 700"
)
# Or use the typed API
from agentflow.agents import PlannerAgent, ExecutorAgent
plan = await PlannerAgent().create_plan(
intent="Sync inventory across all warehouses",
available_apis=orchestrator.discover_apis()
)
result = await ExecutorAgent().execute_plan(plan)Each orchestration is handled by specialized agents (Planner, Executor, Validator) that communicate through a shared context and can negotiate execution strategies.
First-class integration with MuleSoft Anypoint Platform: auto-discovery of APIs from Exchange, RAML/OAS parsing, CloudHub deployment awareness, and runtime policy compliance.
The Dynamic Router scores candidate APIs on latency (P95), cost-per-call, current rate-limit headroom, and semantic capability match — then selects the optimal endpoint in real time.
Adaptive circuit breakers learn from failure patterns. Retry policies adjust backoff based on error classification. Fallback chains provide graceful degradation.
pip install agentflowSee the docs/ directory for detailed guides:
Are you using AgentFlow at your company or in a project? We'd love to hear from you!
👉 Open an Adoption Story issue — takes 2 minutes and helps the project grow.
| Company / Project | Industry | Use Case |
|---|---|---|
| Your company here | Your industry | Share your story → |
| Channel | Purpose |
|---|---|
| 💬 Discussions — Show & Tell | Share what you built |
| ❓ Discussions — Q&A | Ask questions |
| 🔌 Integration Requests | Request a new connector |
| ✨ Feature Requests | Suggest improvements |
| 🐛 Bug Reports | Report issues |
If AgentFlow saves you time or solves a real problem, a ⭐ on this repo goes a long way — it helps more engineers find the framework.
Apache License 2.0 — see LICENSE for details.
Venkata Pavan Kumar Gummadi