Metadata-Version: 2.4
Name: AsyncAgentic
Version: 0.1.2
Summary: Async Agentic Framework For Production Grade Agentic Systems
License-Expression: GPL-3.0-only
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx<1.0.0,>=0.27.0
Requires-Dist: mcp<2.0.0,>=1.28.0
Requires-Dist: openai<3.0.0,>=2.43.0
Requires-Dist: structlog<27.0.0,>=26.1.0
Requires-Dist: tiktoken<1.0.0,>=0.7.0
Description-Content-Type: text/markdown

# AsyncAgentic

Fully Asynchronous No Bloat Python Agentic Framework


## 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:

```bash
uv add AsyncAgentic
```

Install the package via pip:
```bash
pip install AsyncAgentic
```


# USAGE:
## Simple Agent
```python
import os
import json
import asyncio
from datetime import datetime


from AsyncAgentic.Agents import AsyncOpenAISimpleAgent

# 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",
        model="gpt-4o-mini",
        # model="llama3-groq-tool-use",
        api_key=os.environ["OPENAI_API_KEY"],
        context_handling_method="simple",
        max_context_length=25000,
        max_token_per_message=4000,
        max_turns=None,
        max_tool_calls=None,
        debug_print=True,
        # base_url="http://localhost:11434/v1", # OLLAMA ALSO WORK , SO HERE I HAVE ADDED EXAMPLE.
        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:
```json
{
  "stop_reason": "completed",
  "history": {
    "messages": [
      {
        "role": "user",
        "content": "What's the time and weather in Tokyo and London? i am testing the concurrent execution of tools. execute both tools at same time."
      },
      {
        "role": "assistant",
        "content": "The current time is **01:55:00**. \n\nThe weather is:\n- **Tokyo**: Sunny, 22\u00b0C\n- **London**: Sunny, 22\u00b0C"
      },
      {
        "role": "user",
        "content": "And what about New York?"
      },
      {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_KsLd1GublEHZCvoRpTx2nDfE",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"city\":\"New York\"}"
            }
          }
        ]
      },
      {
        "role": "tool",
        "tool_call_id": "call_KsLd1GublEHZCvoRpTx2nDfE",
        "content": "Sunny, 22\u00b0C in New York"
      },
      {
        "role": "assistant",
        "content": "The weather in New York is also **Sunny** with a temperature of **22\u00b0C**."
      }
    ],
    "simplified": [
      {
        "role": "user",
        "content": "What's the time and weather in Tokyo and London? i am testing the concurrent execution of tools. execute both tools at same time."
      },
      {
        "role": "assistant",
        "content": "The current time is **01:55:00**. \n\nThe weather is:\n- **Tokyo**: Sunny, 22\u00b0C\n- **London**: Sunny, 22\u00b0C"
      },
      {
        "role": "user",
        "content": "And what about New York?"
      },
      {
        "role": "assistant",
        "content": "The weather in New York is also **Sunny** with a temperature of **22\u00b0C**."
      }
    ]
  },
  "agent_name": "Test_Agent",
  "run": {
    "turns": 2,
    "tool_calls": 1
  },
  "timestamp": "2025-05-27T01:55:05.698163",
  "output": "The weather in New York is also **Sunny** with a temperature of **22\u00b0C**.",
  "usage": {
    "completion_tokens": 20,
    "prompt_tokens": 186,
    "total_tokens": 206,
    "completion_tokens_details": {
      "accepted_prediction_tokens": 0,
      "audio_tokens": 0,
      "reasoning_tokens": 0,
      "rejected_prediction_tokens": 0
    },
    "prompt_tokens_details": {
      "audio_tokens": 0,
      "cached_tokens": 0
    }
  }
}
```
## Advance Example
here we will use stop chat and event hooks.
```python
import asyncio
import json
from datetime import datetime
import os
import time

from AsyncAgentic.Agents import AsyncOpenAISimpleAgent

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",
        model="gpt-4o-mini",
        api_key=os.environ["OPENAI_API_KEY"],
        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:

```python
import asyncio
import json
from dotenv import dotenv_values
from pydantic import BaseModel, Field
from typing import List
from AsyncAgentic.Agents import AsyncOpenAISimpleAgent
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",
        model=model,
        api_key=api_key,
        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`:

```python
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.
- **model**: LLM model (e.g., `gpt-4o-mini`, `llama3-groq-tool-use`).
- **api_key**: API key for the LLM provider.
- **base_url**: Optional custom endpoint for LLM (e.g., local Ollama server).
- **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.

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:

```python
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.

### MCP HTTP transports

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

```python
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).
- **Retry Strategy**: Configurable retry policies for rate limits, server errors, and timeouts.
- **Streaming Response Agent**: Support for streaming responses from LLMs.
- **Multi-Model Agent**: Support for orchestrating multiple LLMs in a single agent.
- **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.
- **Retry Strategy**: Add configurable retry policies for production reliability.
- **Multi-Model Support**: Enable agents to orchestrate multiple LLMs.
- **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.

Release checks used by this project:

```bash
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.
- Budget control and retry strategies are planned but not implemented due to complexity and dependency concerns.