VenusAI - Advanced Agent Framework 🚀

VenusAI is a secure and extensible Agent framework built for modern AI applications. It offers dynamic tool management, powerful decorators, advanced caching, robust error handling, a built-in CLI, and seamless Claude MCP integration.

---

Installation

Install library via pip or uv.

Note: The venusai is alias of venai, you can use both but venai is the main package.

pip install venai
pip install venusai

or

uv add venai
uv add venusai

Install latest NodeJS with npx for Claude Desktop HTTP support.

Note: mcp-remote package used for support.

---

🔑 Key Capabilities

• 🛡️ Security-first design with permission-based tool filtering & E2B sandbox integration
• 🔧 Dynamic tool ecosystem with decorators for safety, autofix & error recovery
• ⚡ High-performance caching with multiple backends (aiocache, lrucache, async-lru, cachetools)
• 🌐 HTTP API generation → automatically expose tools as REST endpoints
• 🤖 MCP Protocol native support → seamless Claude Desktop integration
• 🎯 Type-safe dependency injection with advanced context management
• 🔄 Self-healing tools → automatic error recovery & retry mechanisms
• 📊 Comprehensive error tracking with detailed frame info & custom handlers

Whether you're building simple chatbots or complex multi-agent systems, VenusAI provides the foundation for scalable, maintainable, and secure AI applications.

---

✨ Features

🔹 Core Bases

• Venus

  • Base class for all Agents
  • No default toolset (bare Agent)

• VenusCode

  • Subclass of Venus with coding capabilities
  • Built-in filesystem toolset
  • Permission-based tool filtering (supports custom permitters)
  • Code execution disabled by default
  • E2B sandbox integration for safe execution

---

🔹 Tools

• Dynamic tool integration from modules

• Dynamic Dependency Injection

• Decorators

  • @agent.safe → error-safe wrapper for non-context tools
  • @agent.safe_plain → error-safe wrapper for context tools
  • @agent.autofix → self-healing tools (functions can fix themselves)
  • @agent.on_error → custom error handler

• Register tools as HTTP endpoints (beta)

  • Convert registered tools to HTTP API (via FastAPI)
  • Just call agent.tools_http_api() and agent.serve()

• Sync/Async caching for tools with @cached

  • Backends: aiocache, lrucache, async-lru, cachetools

• Autofix mechanism

  • Implicitly handles errors via @safe
  • Customizable fix-prompt & fix-model
  • Falls back to a default model if none provided

• Error Handlers

  • Errors yield an ErrorDict with frame & error details
  • Fully customizable responses/actions

---

🔹 Example

from venus import Venus
from venus.errors import ErrorDict
from venus.types import CacheDeps, Deps, DepsT, ModelRetry, RunContext

import hashlib
import logfire

logfire.configure(console=logfire.ConsoleOptions(show_project_link=False))
logfire.instrument_pydantic_ai()

agent = Venus("grok:grok-3", deps_type=int)

class Bank(Deps[DepsT]):
    reserve: int
    """Current bank reserves."""

@agent.on_error
async def retry_on_failure(err: ErrorDict):
    print(f"Error occurred: {err.exception} at {err.location}. Retrying...")
    raise ModelRetry(err.exception)

@agent.on_error
async def notify(err: ErrorDict):
    # e.g: await send_mail(body=err.message)
    pass

def get_reserves():
    return 1_881_938

def get_details():
    return {'code': 'tr', 'swift': 1283, 'id': 1710}

@agent.safe(retries=3, deps=Deps(reserve=get_reserves, details=get_details))
async def add_money(ctx: RunContext[Bank[int]], fund: int):
    if fund <= 5:
        raise ValueError("Enter a number greater than 5.")
    
    ctx.deps.reserve += fund
    bank_details = ctx.deps.get(dict)
    bank_id = bank_details['id']
    tx_hash = hashlib.md5(str(bank_id + ctx.deps.reserve).encode()).hexdigest()
    
    print(f"Connected bank with ID {bank_details['code'].upper()}{bank_details['swift']}")
    print(f"Added ${fund} to current (${ctx.deps.reserve - fund}) reserves.")
    print(f"Hash for transaction: {tx_hash}")
    
    return ctx.deps.reserve

@agent.safe(deps=CacheDeps(id=lambda: 7))
async def test(ctx: RunContext[CacheDeps]):
    return ctx.deps.id

Run:

result = agent.run_sync("Add random money to the bank, pick 4 to 6.", output_type=int)
print(result.output)

or

a2a = agent.to_a2a()

venus serve agent:agent a2a --env dev

✅ This example is complete and runnable as-is.

---

🔹 MCP (Model Context Protocol)

• Tool integration from modules via @tool / @mcp_tool
• Dynamic Claude configuration with MCP.configure(configure_claude=True)
• Dependency Injection support for MCP tools
• mcp-remote integration with HTTP/SSE for Claude Desktop

---

🔹 CLI

Venus provides a command-line interface (CLI) to manage and run agents. You can start chats, serve APIs, or launch MCP servers directly from the terminal.

Available Commands

• Chat with an agent

venus chat module:app

• Run MCP Server

venus mcp --path my_tools.py --name "Venus MCP" --host 127.0.0.1 --port 8000 --transport <sse|http|stdio> --configure

venus mcp --path my_tools.py --name "Venus MCP" --host 127.0.0.1 --port 8000 --transport <sse|http|stdio> --configure --all

• Serve an Agent as API

venus serve mymodule:agent --auto --env dev

CLI Options

• chat → Start interactive CLI chat with an agent
• mcp → Run an MCP server with tools from modules
• serve → Expose your agent via HTTP (FastAPI/Uvicorn)
• Supports plugins such as A2A (a2a)

---

⚡ Usage Examples

Basic Agent

from agent import Venus

agent = Venus(name="venus")
response = agent.run_sync("Hello there!")
print(result.output)

Code-Capable Agent

from venus import VenusCode
from venus.permissions import Permission
from venus.helpers.io import io_toolset

def my_permitter(permission: int):
    if not permission & Permission.EXECUTE and permission & Permission.READ:
        return ["read_file_content"]
    return list(io_toolset.tools.keys())

code_agent = VenusCode(
    name="coder",
    permission=Permission.READ_EXECUTE,
    permitter=my_permitter,  # do not set a permitter to use default permitter
)

Dependency Injection

from venus import Venus
from venus.types import Deps, DepsT, RunContext

import uuid
import time

agent = Venus(deps_type=int)

uuidgen = lambda: uuid.uuid4().hex
datagen = lambda: {'foo': [Deps(bar='baz')]}

class Auth(Deps[DepsT]):
    id: str

@agent.safe(deps=Deps(id=uuidgen, data=datagen))
def get_tx(ctx: RunContext[Auth[int]]): # AgentDepsT is int here
    # attribute-style access to deps entity `id`
    txhash = f'%d$%s' % (time.time(), ctx.deps.id)
     # type-based access to deps entity `foo`
    data = ctx.deps.get(dict) # None
    data = ctx.deps.get(list) # [Deps(bar='baz')]

    # access main dependency for agent
    agentdeps = ctx.deps.main # int
    
    # type-based access to deps entity `foo`
    # use exact annotation to access it:
    data = ctx.deps.get(list[Deps]) # [Deps(bar='baz')]
    return txhash + data.bar

Module Tools with Decorators

# agent.py
from venus import Venus
agent = Venus(tool_modules='agent_tools')

# agent_tools.py
from venus.types import Deps
from venus.caching import cached
from venus.decorators import tool, mcp_tool, safe_call, autofix

@tool
@cached(ttl=240)
def get_id():
    return 1

@mcp_tool(deps=Deps(id=get_id))
def get_username(deps: Deps):
    return f'@user{deps.id}'

@safe_call
async def create_user(username: str):
    return True

@autofix
async def risky_function():
    raise Exception('An error occured')

Agent Tools with Decorators

# agent.py
from venus import Venus
from venus.types import RunContext

agent = Venus()

@agent.safe_plain
def add(x: int, y: int) -> int:
    return x + y

@agent.safe(retries=3)
def sub(ctx: RunContext, x: int, y: int) -> int:
    return x - y

@agent.autofix(retries=2, deps=Deps(result=lambda: 20))
def risky_function(data: str):
    raise Exception('An error occured')

---

🛠 Tech Stack

• Python 3.10+ → async-first with modern type hints
• Based on PydanticAI → robust validation & AI agent foundation
• ASGI-compatible → works with FastAPI, Uvicorn, etc.
• MCP Protocol → native Model Context Protocol integration
• Secure execution with E2B Sandbox
• CLI powered by Click → ergonomic, extensible command line
• Advanced Caching → multiple backend support
• Dependency Injection → type-safe, dynamic DI system
• Error Handling → custom recovery & retry strategies
• Decorator System → tool safety, autofix & error control
• HTTP API Generation → auto REST endpoint conversion

---

🤝 Contributing

Contributions are welcome! 🎉 Please open an issue before submitting a PR to discuss your idea.

---

📜 License

Licensed under the MIT License – see the LICENSE file for details.