AsyncAgentic

Fully Asynchronous No Bloat Python Agentic Framework

[!IMPORTANT]

LLM-assisted development disclaimer: I have used LLM assistance in development and maintainance of this project.

Overview

AsyncAgentic is a lightweight, production-oriented, fully asynchronous Python framework for building agentic systems powered by OpenAI and OpenAI-compatible model APIs. It is designed to prioritize simplicity, extensibility, and performance, allowing developers to focus on business logic without wrestling with complex framework abstractions. The framework supports concurrent tool execution and provides controls such as event hooks and stop signals.

Key principles:

• No Bloat: Minimal dependencies, only what's necessary.
• Fully Asynchronous: No blocking calls, leveraging Python's asyncio for performance.
• Simple to Use: Intuitive API, no steep learning curve.
• Extensible and Debuggable: Easy to extend with custom logic and debug with detailed logs.
• Production Controls: Features like user_id/chat_id, stop signals, bounded loops, and event hooks for integration into larger systems.
• No Code Execution: The framework does not execute LLM-generated code.
• Direct Tool Execution: No human-in-the-middle; tool calls are executed directly, and only LLM text responses are returned to the user.

Features

• No Bloat: Minimal dependencies to keep the framework lightweight.
• Fully Asynchronous: Built with asyncio for non-blocking operations.
• Simple API: Intuitive interface for quick integration.
• Parallel Tool Calls: Supports concurrent execution of multiple tool calls for efficiency.
• Event Hooks: Customizable hooks for monitoring and extending functionality:
  • on_function_call_start: Triggered when a tool call begins.
  • on_function_call_end: Triggered when a tool call completes.
  • on_function_call_error: Triggered on tool call errors.
  • on_context_overflow: Triggered when context limits are exceeded.
  • on_message_dropped: Triggered when an atomic context unit is removed.
  • on_max_turns_reached and on_max_tool_calls_reached: Triggered at safety limits.
  • on_manual_stop, on_final_response, and on_error: Observe terminal events.
• Stop Chat System: Attach a stop signal function to halt chat execution dynamically.
• Context Management: Choose between "Simple" (length-based) or "Accurate" (token-based with tiktoken) context handling.
• Forced User and Chat IDs: Every function and hook receives user_id and chat_id for production-grade tracking and integration (e.g., cost tracking, user-specific logic).
• No Human-in-the-Middle: Tool calls are executed directly without exposing internal workings to users.
• Customizable System Prompts: Tailor agent behavior with system prompts.
• Flexible Tool Registry: Register tools with JSON schemas compatible with OpenAI's format.
• Concurrent Function Execution: Execute multiple tool calls simultaneously when enabled.
• Optional Agent Loop Limits: Supports opt-in model-turn and tool-call limits and reports why a limited run stopped.
• Safe Context Pruning: Keeps the system prompt and latest user request, and never separates a tool call from its output.
• Per-Tool Execution Policies: Keep reads concurrent while serializing writes globally or by a business-defined key.
• Stable API: Version 1 APIs will be final with no breaking changes, ensuring long-term support (LTS).

Installation

AsyncAgentic requires Python 3.10 or newer.

Install the package via uv:

uv add AsyncAgentic

Install the package via pip:

pip install AsyncAgentic

USAGE:

Simple Agent

import os
import json
import asyncio
from datetime import datetime


from AsyncAgentic.Agents import AsyncOpenAISimpleAgent, LLMProviderConfig

# NOTICE: FORCED DEPENDECY? CHAT_ID AND USER_ID IS COMPULSORY. 
# THIS MAY FEEL WEIRD BUT YOU WILL THANK ME LATER. WHEN YOUR PRODUCTION REQUIREMENT CHANGES. LIKE AGENT SPECIFIC COSTING OR ONLY PURCHASED AGENTS ARE ACCESIBLE ETC...
async def get_current_time(user_id: str, chat_id: str, agent_name: str) -> str:
    print(f"get_current_time called by {agent_name} for user {user_id} and chat {chat_id}")
    return datetime.now().strftime("%H:%M:%S")

async def get_weather(city: str, user_id: str, chat_id: str, agent_name: str) -> str:
    print(f"get_weather called by {agent_name} for user {user_id} and chat {chat_id}")
    await asyncio.sleep(1)
    return f"Sunny, 22°C in {city}"

get_time_schema = {
    "name": "get_current_time",
    "description": "Get the current time",
    "parameters": {
        "type": "object",
        "properties": {},
        "required": []
    }
}

get_weather_schema = {
    "name": "get_weather",
    "description": "Get weather for a city",
    "parameters": {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "The city to get weather for"
            }
        },
        "required": ["city"]
    }
}

async def main():
    agent = AsyncOpenAISimpleAgent(
        agent_name="Test_Agent",
        agent_description="Test agent for weather and time",
        llm_provider_config_list=[
            LLMProviderConfig(
                name="deepseek-primary",
                model="deepseek-chat",
                api_key=os.environ["DEEPSEEK_API_KEY"],
                provider_type="openai",
                base_url="https://api.deepseek.com/v1",
                retry_amount=2,
                retry_backoff=1.0,
                cooldown_seconds=60.0,
                max_cooldown_seconds=900.0,
                pricing=None,  # Optional LLMProviderPricing for cost estimates.
            ),
            LLMProviderConfig(
                name="openai-fallback",
                model="gpt-4o-mini",
                api_key=os.environ["OPENAI_API_KEY"],
                provider_type="openai",
                base_url="https://api.openai.com/v1",
                retry_amount=2,
                retry_backoff=1.0,
                cooldown_seconds=60.0,
                max_cooldown_seconds=900.0,
                pricing=None,
            ),
        ],
        context_handling_method="simple",
        max_context_length=25000,
        max_token_per_message=4000,
        max_turns=None,
        max_tool_calls=None,
        debug_print=True,
        user_id="test_user",
        chat_id="test_chat",
        tool_registry=[
            {
                "name": "get_current_time",
                "function_schema": get_time_schema,
                "func": get_current_time
            },
            {
                "name": "get_weather",
                "function_schema": get_weather_schema,
                "func": get_weather
            }
        ],
        execute_function_concurrently=True,
        system_prompt="You are a helpful assistant that can check time and weather"
    )


    response = await agent.send_message(
        "What's the time and weather in Tokyo and London? i am testing the concurrent execution of tools. execute both tools at same time.",
        debug_print=True
    )
    print(json.dumps(response, indent=2))

    # TEST 2: CONVERSATION WITH HISTORY , 
    # NOTE: THIS IS JUST TO SHOW YOU GUYS HOW IT WORKS. 

    print("\nTesting conversation with history...")
    response = await agent.send_message(
        "And what about New York?",
        history=response["history"]["simplified"],
    )
    print(json.dumps(response, indent=2))
    await agent.close()

if __name__ == "__main__":
    asyncio.run(main()) 

RESPONSE:

{
  "request_id": "request_...",
  "stop_reason": "completed",
  "history": {
    "messages": [
      {
        "item": {"role": "user", "content": "What's the weather?"},
        "metadata": {
          "source": "user",
          "request_id": "request_...",
          "timestamp": "..."
        }
      },
      {
        "item": {"type": "message", "content": [{"type": "text", "text": "Sunny."}]},
        "metadata": {
          "source": "llm",
          "request_id": "request_...",
          "model_turn_id": "turn_...",
          "provider_name": "deepseek-primary",
          "model": "deepseek-chat",
          "attempt": 1,
          "timestamp": "...",
          "usage": {
            "input_tokens": 20,
            "cached_input_tokens": 0,
            "output_tokens": 4,
            "reasoning_tokens": 0,
            "total_tokens": 24
          }
        }
      }
    ],
    "simplified": ["same envelope shape with role/content items"]
  },
  "agent_name": "Test_Agent",
  "run": {"turns": 1, "tool_calls": 0},
  "timestamp": "...",
  "output": "Sunny.",
  "usage": {
    "input_tokens": 20,
    "cached_input_tokens": 0,
    "output_tokens": 4,
    "reasoning_tokens": 0,
    "total_tokens": 24
  },
  "provider_usage": [
    {
      "provider_name": "deepseek-primary",
      "model": "deepseek-chat",
      "model_requests": 1,
      "usage": {
        "input_tokens": 20,
        "cached_input_tokens": 0,
        "output_tokens": 4,
        "reasoning_tokens": 0,
        "total_tokens": 24
      },
      "estimated_cost": null
    }
  ],
  "estimated_cost": null,
  "request_trace": [{"provider_name": "deepseek-primary", "action": "success"}]
}

Advance Example

here we will use stop chat and event hooks.

import asyncio
import json
from datetime import datetime
import os
import time

from AsyncAgentic.Agents import AsyncOpenAISimpleAgent, LLMProviderConfig

async def get_current_time(user_id: str, chat_id: str, agent_name: str) -> str:
    print(f"get_current_time called by {agent_name} for user {user_id} and chat {chat_id}")
    return datetime.now().strftime("%H:%M:%S")

async def get_weather(city: str, user_id: str, chat_id: str, agent_name: str) -> str:
    print(f"get_weather called by {agent_name} for user {user_id} and chat {chat_id}")
    return f"Sunny, 22°C in {city}"

get_time_schema = {
    "name": "get_current_time",
    "description": "Get the current time",
    "parameters": {
        "type": "object",
        "properties": {},
        "required": []
    }
}

get_weather_schema = {
    "name": "get_weather",
    "description": "Get weather for a city",
    "parameters": {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "The city to get weather for"
            }
        },
        "required": ["city"]
    }
}

# DEFINING STOP SIGNAL FUNCTION. I AM USING SIMPLE SLEEP HERE TO SHOW EXAMPLE.
# YOU CAN USE REDIS IN TERMS OF DISTRIBUTED SYSTEMS.
# OR HOWEVER YOU WANT IT, BASICALLY YOUR FUNCTION MUST RETURN TRUE IF YOU WANT TO STOP CHAT AT ANY POINT.
# DO NOTE: THIS FUNCTION IS NOT ACTUAL REPRESENTATION OF STOP SYSTEM YOU MAKE IN PRODUCTION.
# HERE ANY PROCESS WHICH IS TAKING MORE THAN 3 SECONDS TO COMPLETE WILL BE STOPPED. BUT IN PROD YOU CAN TRIGGER THIS MANUALLY.

async def stop_chat_signal(user_id: str, chat_id: str) -> bool:
    print(f"stop_chat_signal called for user {user_id} and chat {chat_id}")
    # you will have chat_id and user_id based listner in your system
    await asyncio.sleep(3) # so agents might be running but after 3 secounds in , it will stop at that place. 

    print(f"stop_chat_signal is returning True at {datetime.now()}")
    return True

async def hook_on_function_call_end(data):
    print(f"THIS MESSAGE IS FROM HOOK ON FUNCTION CALL END: {data}")

async def main():
    agent = AsyncOpenAISimpleAgent(
        agent_name="Test_Agent",
        agent_description="Test agent for weather and time",
        llm_provider_config_list=[
            LLMProviderConfig(
                name="openai-primary",
                model="gpt-4o-mini",
                api_key=os.environ["OPENAI_API_KEY"],
                provider_type="openai",
            )
        ],
        context_handling_method="simple",
        max_context_length=25000,
        max_token_per_message=4000,
        user_id="test_user",
        chat_id="test_chat",
        tool_registry=[
            {
                "name": "get_current_time",
                "function_schema": get_time_schema,
                "func": get_current_time
            },
            {
                "name": "get_weather",
                "function_schema": get_weather_schema,
                "func": get_weather
            }
        ],
        execute_function_concurrently=True,
        system_prompt="You are a helpful assistant that can check time and weather",
        manual_stop_signal_function=stop_chat_signal,
        hooks={
            'on_function_call_end': hook_on_function_call_end
        }
    )
    start_time = time.perf_counter()
    print(f"STARTING CHAT AT {datetime.now()}")
    print("\nTesting multiple tool calls... & stop chat & event hooks")
    response = await agent.send_message(
        "What's the time and weather in Tokyo and London? i am testing the concurrent execution of tools. execute both tools at same time.",
        debug_print=True
    )
    print(json.dumps(response, indent=2))

    print("\nTesting conversation with history...")
    response = await agent.send_message(
        "And what about New York, USA , France , Germanay, Florida?",
        history=response["history"]["simplified"],
        debug_print=True,
    )
    print(json.dumps(response, indent=2))
    end_time = time.perf_counter()
    print(f"Total time taken: {end_time - start_time} seconds")

if __name__ == "__main__":
    asyncio.run(main()) 

MCP Integration (BrowserOS)

AsyncAgentic supports the Model Context Protocol (MCP) out of the box, allowing you to connect agents to external tools and environments like BrowserOS.

Here is a complete example of connecting to the BrowserOS MCP server to list open tabs and inspect their states:

import asyncio
import json
from dotenv import dotenv_values
from pydantic import BaseModel, Field
from typing import List
from AsyncAgentic.Agents import AsyncOpenAISimpleAgent, LLMProviderConfig
from AsyncAgentic import configure_logging

# Configure logging at INFO level to see tool executions
configure_logging(level="INFO")

# Define Pydantic models for structured output parsing in client scripts
class TabInfo(BaseModel):
    name: str = Field(description="The title/name of the browser tab")
    url: str = Field(description="The URL of the browser tab")
    description: str = Field(description="A brief description of what the tab is about")

class TabsList(BaseModel):
    tabs: List[TabInfo] = Field(description="List of active/existing browser tabs")

async def main():
    # Load configuration
    envtokens = dotenv_values("../asenv/.env")
    model = envtokens.get("LLM_MODEL") or "gpt-4o"
    api_key = envtokens.get("LLM_API_KEY")
    base_url = envtokens.get("LLM_BASE_URL") or "https://api.openai.com/v1"
    if not api_key:
        raise RuntimeError("Set LLM_API_KEY before running this example")

    # Initialize the agent with BrowserOS MCP server config
    agent = AsyncOpenAISimpleAgent(
        agent_name="BrowserOS_Agent",
        agent_description="Agent equipped with local BrowserOS MCP",
        llm_provider_config_list=[
            LLMProviderConfig(
                name="primary",
                model=model,
                api_key=api_key,
                provider_type="openai",
                base_url=base_url,
            )
        ],
        user_id="browser_user",
        chat_id="browser_chat",
        system_prompt="You are a helpful assistant. Use tools to check browser tabs.",
        mcp_servers={
            "browseros": {
                "command": "npx",
                "args": ["mcp-remote", "http://127.0.0.1:9200/mcp"]
            }
        }
    )

    async with agent:
        response = await agent.send_message(
            "Please list all existing browser tabs on my running browser using the BrowserOS tools."
        )

        print("\n--- Agent Response ---")
        print(response.get("output", ""))

if __name__ == "__main__":
    asyncio.run(main())

Logging Configuration

AsyncAgentic uses structlog for structured logging. By default, it pretty-prints logs to standard output. You can customize the log level and format using configure_logging:

from AsyncAgentic import configure_logging

# Configure pretty-printed console logging with DEBUG level
configure_logging(level="DEBUG")

# Configure JSON format for production environments
configure_logging(level="INFO", json_format=True)

Configuration Options

• agent_name: Unique identifier for the agent.
• user_id / chat_id: Required keyword-only identifiers injected into every local tool and hook.
• llm_provider_config_list: Required ordered list of typed LLMProviderConfig objects. The first available provider is preferred and later entries are fallbacks.
• context_handling_method: simple (length-based) or Accurate (token-based with tiktoken).
• max_context_length: Optional total context limit in tokens. Defaults to None; configure it for the selected model when framework-side pruning is desired.
• max_token_per_message: Optional per-message truncation limit. Defaults to None, so tool and browser results are not silently shortened.
• max_messages_in_context: Optional message-count limit. Defaults to None; total-token limits are usually a better control for long agent workflows.
• prompt_when_context_overflow: Custom prompt for context overflow scenarios.
• prompt_when_message_is_dropped: Custom prompt when messages are dropped.
• max_turns: Optional maximum model requests in one agent run. Defaults to None so long research agents are not stopped artificially; set an integer when a workflow needs a hard budget.
• max_tool_calls: Optional maximum client-executed local/local-MCP calls in one run. Defaults to None. OpenAI-hosted native MCP calls occur inside the Responses API and are not included in this client-side count.
• tool_registry: List of tools with their schemas and functions.
• execute_function_concurrently: Enable concurrent execution of multiple tool calls.
• manual_stop_signal_function: Custom function to signal chat termination.
• hooks: Dictionary of event hooks for custom logic.

Providers, retry health, and costing

Every agent owns its provider clients and health state. State is not shared between agents. A 429 places that provider in cooldown (respecting Retry-After) and new requests on the same agent skip it until a half-open probe succeeds. Authentication, payment, permission, and missing-model failures disable only that agent's provider route until await agent.reset_llm_provider_health(name) is called. Inspect a redacted snapshot with await agent.get_llm_provider_health(). SDK retries are disabled so retry_amount and hook events match real network attempts.

from decimal import Decimal
from AsyncAgentic import LLMProviderConfig, LLMProviderPricing

providers = [
    LLMProviderConfig(
        name="primary",
        model="provider-model",
        api_key="...",
        provider_type="openai",
        base_url="https://provider.example/v1",
        retry_amount=2,
        retry_backoff=1.0,
        cooldown_seconds=60,
        pricing=LLMProviderPricing(
            input_per_million=Decimal("0.50"),
            output_per_million=Decimal("1.50"),
        ),
    ),
    LLMProviderConfig(
        name="fallback",
        model="fallback-model",
        api_key="...",
        provider_type="openai",
        base_url="https://fallback.example/v1",
    ),
]

provider_type is typed as Literal["openai", "gemini"] and defaults to "openai". Use "openai" for the standard OpenAI-compatible wire format, including compatible third-party providers. Set it explicitly to "gemini" for a Gemini route:

gemini_provider = LLMProviderConfig(
    name="gemini-primary",
    model="gemini-3-flash-preview",
    api_key="...",
    provider_type="gemini",
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
)

Gemini routes preserve tool-call thought signatures. When history comes from a different provider and has no Gemini signature, AsyncAgentic applies Gemini's cross-provider compatibility marker to the first tool call in that assistant step. Provider behavior is never inferred from name, model, or base_url.

retry_amount counts retries after the initial attempt. Rate limits fall through immediately instead of repeatedly attacking a limited provider; connection errors, timeouts, 408, 409, and 5xx use exponential backoff before fallback.

Usage is normalized across Responses and Chat Completions. Optional pricing produces per-provider and total estimated_cost values; these are estimates rather than provider invoices.

Typed history and provider attribution

History entries use {"item": ..., "metadata": ...} envelopes. item contains the portable OpenAI-compatible conversation item. metadata records the source, request/model-turn IDs, provider, model, normalized usage, and estimated cost. AsyncAgentic strips metadata before every provider call. Failed attempts are kept in request_trace, never inserted into conversational history. Both messages and simplified histories can be passed directly into the next send_message() call.

Use the agent as an async context manager, or call await agent.close() when it is no longer needed. Closing releases local MCP processes/sessions and the model HTTP connection pool. A closed agent cannot be reused.

Tool execution policies

execute_function_concurrently=False keeps an entire tool-call batch sequential. When batch concurrency is enabled, an individual registry entry may still request stricter execution:

tool_registry = [
    {
        "name": "update_post",
        "function_schema": update_post_schema,
        "func": update_post,
        "execution": "locked",
        "lock_key": lambda arguments: f"post:{arguments['post_id']}",
    },
    {
        "name": "create_post",
        "function_schema": create_post_schema,
        "func": create_post,
        "execution": "sequential",
    },
]

concurrent is the default. sequential serializes calls carrying that policy. locked serializes calls that resolve to the same lock_key, while different keys can still run concurrently. The callable receives only model-supplied arguments; user_id, chat_id, and agent_name remain framework-injected.

Tool timeouts intentionally belong to tool implementations, where the appropriate HTTP/client timeout is known. Use asyncio.timeout() inside a tool when a hard deadline is needed.

Structured tool results

Local tools may return strings, dictionaries, lists, dataclasses, or Pydantic models. Non-string results are encoded as JSON before they are returned to the model, so a structured WordPress result such as {"post_id": 42, "status": "updated"} remains structured. MCP structuredContent and MCP error status are preserved as well.

Hook event data

Tool hooks include function, call_id, arguments, source, user_id, chat_id, and agent_name. Completion/error hooks additionally include start and completion timestamps, duration, and either result or error. Hook failures are logged and never replace the agent result.

Model routing hooks are on_model_request_start, on_model_request_success, on_model_request_error, on_model_retry, on_provider_fallback, and on_provider_unavailable. They include request/model-turn IDs, provider name and index, model, attempt, timing, status/action, and normalized usage when available. API keys are never included.

MCP HTTP transports

URL-based local MCP servers default to the legacy sse transport for backward compatibility. Streamable HTTP is available explicitly:

mcp_servers={
    "research": {
        "url": "https://example.com/mcp",
        "transport": "streamable_http",
    }
}

Native OpenAI MCP/connectors require the Responses API. If an OpenAI-compatible provider only supports Chat Completions, AsyncAgentic raises a clear error rather than silently dropping those native tools. Client-side/local MCP tools continue to work through the Chat Completions fallback.

Because this framework intentionally has no human-in-the-loop approval UI, every entry in native_mcp_servers must explicitly set "require_approval": "never". AsyncAgentic rejects the configuration otherwise. Only use that setting with MCP servers and individual tools you have reviewed and trust. Local/client-side MCP servers are executed directly by your process and do not use OpenAI's approval flow.

Planned Features

• Budget Control System: Per-chat budget limits with hooks for exceeding budgets (pending OpenAI pricing API).
• Streaming Response Agent: Support for streaming responses from LLMs.
• GUI Management: Optional GUI for managing functions and agents (under consideration, may not be implemented to avoid bloat).

Work Left

• Context Handling: Improve context management with advanced strategies (e.g., summary-based dropping).
• Image Tools: Support for processing and generating images.
• Error Handler: Robust error handling for tool calls and API interactions.
• Data Models: Data models for event hooks and stop signals.

Roadmap

• Version 1.0 (LTS): Stable release with finalized APIs, no breaking changes.
• Context Management Enhancements: Advanced strategies like summary-based dropping and UI integration for context visualization.
• Budget Control: Implement per-chat budget limits with hooks once OpenAI pricing API is available.
• Streaming Support: Add streaming response capabilities for real-time interactions.
• Documentation Expansion: Detailed guides for production use cases, hooks, and stop signals.

Contributing

Contributions are welcome! Please submit issues or pull requests to the GitHub repository. Focus on maintaining simplicity and avoiding unnecessary dependencies.

License

AsyncAgentic is available under the permissive MIT License.

Release checks used by this project:

ruff check src tests
npx pyright
uv run python -m unittest discover -s tests -v
uv build

Notes

• Do not use this framework in production till V1.0.0 is Released.
• The framework enforces user_id and chat_id for all functions and hooks to enable production-grade tracking (e.g., cost management, user-specific logic).
• For production systems, rely on direct tool execution to maintain transparency and control.