Compare commits

..
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison ed90426d96 docs: enhance CONTRIBUTING.md with setup instructions and testing guidance
- Add detailed installation steps with pip install commands
- Specify PEP 8 as code style standard
- Include pytest test running instructions
- Add guidance to check/create issues for large changes
- Include link to GitHub Issues for support

Addresses feedback from code review bots (CodeRabbit, Gemini, Greptile)

Co-authored-by: Mervin Praison <[email protected]>
2026-05-09 18:34:13 +00:00
Akhil Chalasani 9c32b9d84c docs: improve documentation
Created CONTRIBUTING.md
2026-05-09 13:22:09 -05:00
321 changed files with 8674 additions and 22693 deletions

No files matched your search

-51
View File
@@ -1,51 +0,0 @@
# PraisonAI Environment Variables
# Copy this file to .env and update with your actual values
# === Required ===
OPENAI_API_KEY=sk-... # OpenAI key for LLM calls. Get yours at https://platform.openai.com
# === Required for Claw web-search tool ===
TAVILY_API_KEY=tvly-... # Get yours at https://app.tavily.com (free tier available)
# === Optional: Claw dashboard security ===
PRAISONAI_ALLOW_LOCAL_TOOLS=0 # Set to 1 to enable local shell tools (security risk)
# === Optional: Alternative LLM Providers ===
ANTHROPIC_API_KEY=sk-ant-... # Anthropic Claude models
GOOGLE_API_KEY=... # Google Gemini models
GROQ_API_KEY=gsk_... # Groq (fast inference)
COHERE_API_KEY=... # Cohere models
AZURE_OPENAI_API_KEY=... # Azure OpenAI
AZURE_OPENAI_ENDPOINT=... # Azure OpenAI endpoint
# === Optional: Memory backends ===
MEM0_API_KEY=... # Mem0 memory service
REDIS_URL=redis://localhost:6379 # Redis for state management
# === Optional: Observability ===
LANGFUSE_SECRET_KEY=sk-lf-... # Langfuse tracing
LANGFUSE_PUBLIC_KEY=pk-lf-... # Langfuse public key
# === Optional: Bot platforms ===
TELEGRAM_BOT_TOKEN=... # Telegram bot token
DISCORD_BOT_TOKEN=... # Discord bot token
SLACK_BOT_TOKEN=xoxb-... # Slack bot token
SLACK_APP_TOKEN=xapp-... # Slack app token
WHATSAPP_ACCESS_TOKEN=... # WhatsApp bot access token
WHATSAPP_PHONE_NUMBER_ID=... # WhatsApp phone number ID
# === Optional: Database ===
DATABASE_URL=sqlite:///~/.praison/database.sqlite # Database connection string
# === Optional: Tool Security ===
# ALLOWED_TOOLS=search,send_message # Whitelist specific tools to prevent name collisions
# Useful in multi-environment setups with overlapping tool names
# Leave unset to use all tools (with collision warnings)
# Do NOT set to empty string: that's treated as configuration error
# In CI, unknown tool names in the whitelist cause strict startup failure
# HERMES_ONLY_TOOLS=search,send_message # (Deprecated) Backward compatibility alias for ALLOWED_TOOLS
# === Optional: Development ===
LOGLEVEL=INFO # Logging level (DEBUG, INFO, WARNING, ERROR)
CHAINLIT_AUTH_SECRET=... # Chainlit authentication secret
DEBUG=false # Enable debug mode
+2 -5
View File
@@ -181,15 +181,12 @@ jobs:
- name: Fetch PR branch and Setup Remote
if: github.event.issue.pull_request
env:
PR_BRANCH: ${{ steps.check_fork.outputs.pr_branch }}
IS_FORK: ${{ steps.check_fork.outputs.is_fork }}
run: |
# Fetch PR head from base repo and put it into local branch
git fetch origin pull/${{ github.event.issue.number }}/head:"${PR_BRANCH}"
git fetch origin pull/${{ github.event.issue.number }}/head:${{ steps.check_fork.outputs.pr_branch }}
# If it's a fork, make origin point to local to trick claude-code-action's `git fetch origin <branch>`
if [ "${IS_FORK}" = "true" ]; then
if [ "${{ steps.check_fork.outputs.is_fork }}" == "true" ]; then
git remote set-url origin file://$(pwd)
fi
@@ -1,158 +0,0 @@
name: Cross-Repo Integration Tests
on:
push:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
pull_request:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
workflow_dispatch:
jobs:
cross-repo-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
# Install PraisonAI packages in development mode
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
# Install PraisonAIUI in development mode
cd ../../../PraisonAIUI
pip install -e .
- name: Verify versions
run: |
python -c "import praisonaiagents; print(f'praisonaiagents: {praisonaiagents.__version__}')"
python -c "import praisonai; print(f'praisonai: {praisonai.__version__}')"
python -c "import praisonaiui; print(f'praisonaiui: {praisonaiui.__version__}')"
- name: Run PraisonAI integration tests
run: |
cd PraisonAI/src/praisonai
python -m pytest tests/integration/test_aiui_* -v --timeout=300
- name: Run PraisonAIUI integration tests
run: |
cd PraisonAIUI
if [ -f tests/integration/test_agentic_roundtrip.py ]; then
python -m pytest tests/integration/test_agentic_roundtrip.py -v --timeout=300
fi
if [ -f tests/test_feature_sdk_backends.py ]; then
python -m pytest tests/test_feature_sdk_backends.py -v --timeout=300
fi
- name: Test Pattern C CLI
run: |
cd PraisonAI/src/praisonai
# Test that ui-gateway command exists
python -m praisonai serve --help | grep ui-gateway
- name: Test public API exports
run: |
cd PraisonAI/src/praisonai
python -c "from praisonai import run_integrated_gateway; print('✓ run_integrated_gateway imported')"
python -c "from praisonai import configure_host; print('✓ configure_host imported')"
- name: Integration smoke test
env:
PRAISONAI_TEST_MODE: "1"
PYTHONPATH: ${{ github.workspace }}/PraisonAI/src/praisonai:${{ github.workspace }}/PraisonAI/src/praisonai-agents:${{ github.workspace }}/PraisonAIUI
run: |
cd PraisonAI/src/praisonai
python -c "
import sys
sys.path.insert(0, '.')
from praisonai.integration.host_app import configure_host
from praisonai import run_integrated_gateway
print('✓ Integration imports successful')
# Test configure_host with new parameters
try:
configure_host(
style='dashboard',
context_paths=['AGENTS.md'],
title='Test App'
)
print('✓ configure_host with new parameters works')
except Exception as e:
print(f'× configure_host failed: {e}')
sys.exit(1)
"
# Optional job with API key for agentic tests (if secret is available)
agentic-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
if: ${{ vars.RUN_AGENTIC_TESTS == 'true' }}
needs: cross-repo-integration
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
cd ../../../PraisonAIUI
pip install -e .
- name: Run agentic integration test
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
cd PraisonAI/src/praisonai
if [ -n "$OPENAI_API_KEY" ]; then
python -m pytest tests/integration/test_aiui_host_agentic.py -v --timeout=600
else
echo "OPENAI_API_KEY not available, skipping agentic tests"
fi
+5 -4
View File
@@ -29,6 +29,9 @@ jobs:
uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
with:
fetch-depth: 0
- name: Generate GitHub App Token
@@ -60,12 +63,10 @@ jobs:
GH_TOKEN: ${{ steps.generate_token.outputs.token }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
PRAISONAI_AUTO_APPROVE: "true"
ISSUE_NUMBER_INPUT: ${{ inputs.issue_number }}
ISSUE_NUMBER_EVENT: ${{ github.event.issue.number }}
run: |
if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
export ISSUE_NUMBER="$ISSUE_NUMBER_INPUT"
export ISSUE_NUMBER="${{ inputs.issue_number }}"
else
export ISSUE_NUMBER="$ISSUE_NUMBER_EVENT"
export ISSUE_NUMBER="${{ github.event.issue.number }}"
fi
praisonai github triage --issue $ISSUE_NUMBER --agent-file .github/praisonai-issue-triage.yaml
+6 -7
View File
@@ -30,23 +30,22 @@ jobs:
steps:
- name: Determine checkout ref
id: dest
env:
ISSUE_NUMBER: ${{ github.event.issue.number }}
PR_NUMBER_INPUT: ${{ inputs.pr_number }}
PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }}
run: |
if [ "${{ github.event_name }}" = "issue_comment" ]; then
echo "ref=refs/pull/$ISSUE_NUMBER/head" >> "$GITHUB_OUTPUT"
echo "ref=refs/pull/${{ github.event.issue.number }}/head" >> "$GITHUB_OUTPUT"
elif [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
echo "ref=refs/pull/$PR_NUMBER_INPUT/head" >> "$GITHUB_OUTPUT"
echo "ref=refs/pull/${{ inputs.pr_number }}/head" >> "$GITHUB_OUTPUT"
else
echo "ref=$PR_HEAD_SHA" >> "$GITHUB_OUTPUT"
echo "ref=${{ github.event.pull_request.head.sha }}" >> "$GITHUB_OUTPUT"
fi
- name: Checkout Repository
uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
with:
ref: ${{ steps.dest.outputs.ref }}
fetch-depth: 0
+2 -14
View File
@@ -26,8 +26,8 @@ jobs:
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
with:
persist-credentials: false
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
@@ -176,18 +176,6 @@ jobs:
run: praisonai queue list
continue-on-error: true
# ==================== PACKAGING TESTS ====================
- name: Test packaging/daemon entry points
run: |
echo "Testing both entry point methods..."
python -m praisonai --version
praisonai --version
- name: Test doctor packaging command
run: |
echo "Testing doctor packaging diagnostics..."
praisonai doctor packaging --json
# ==================== REAL API TESTS ====================
- name: Real API test - Agent run
if: env.OPENAI_API_KEY != ''
+29
View File
@@ -0,0 +1,29 @@
# Contributing to PraisonAI
Thank you for your interest! Here's how to get started:
## Getting Started
1. Fork the repository
2. Clone your fork and install dependencies:
```bash
git clone https://github.com/your-username/PraisonAI.git
cd PraisonAI
pip install -e src/praisonai-agents/
pip install -e src/praisonai/
```
3. Create a feature branch: `git checkout -b feature/your-feature`
4. Make your changes
5. Submit a pull request
## Guidelines
- Follow existing code style (PEP 8)
- Add tests for new features and run them locally using pytest:
```bash
cd src/praisonai-agents && pytest tests/
```
- Update documentation as needed
- Check existing issues or open a new one to discuss your idea before implementing large changes
For questions or support, please visit our [GitHub Issues](https://github.com/MervinPraison/PraisonAI/issues).
+3 -40
View File
@@ -294,35 +294,8 @@ def search(query: str) -> str:
@tool
def calculate(expression: str) -> float:
"""Safely evaluate a numeric arithmetic expression."""
import ast
import operator
# Define allowed operations
_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
def _safe_eval(node):
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
elif isinstance(node, ast.BinOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
elif isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_safe_eval(node.operand))
else:
raise ValueError("Unsupported expression")
try:
return _safe_eval(ast.parse(expression, mode="eval").body)
except (ValueError, SyntaxError, TypeError, ZeroDivisionError, OverflowError):
raise ValueError("Invalid arithmetic expression")
"""Evaluate a math expression."""
return eval(expression)
agent = Agent(
instructions="You are a helpful assistant",
@@ -331,7 +304,6 @@ agent = Agent(
agent.start("Search for AI news and calculate 15*4")
```
> ⚠️ **Security Note:** Never use `eval()`, `exec()`, or `subprocess` in tool functions that process LLM-generated or user-supplied input. Always validate and sanitize inputs to prevent code injection attacks.
> 📖 [Full tools docs](https://docs.praison.ai/docs/tools/tools) — BaseTool, tool packages, 100+ built-in tools
### 5. Persistence (Databases)
@@ -358,15 +330,6 @@ pip install "praisonai[claw]"
praisonai claw
```
#### Required Environment Variables
Copy `.env.example` to `.env` and configure the following variables:
| Variable | Required | Description |
|----------|----------|-------------|
| `OPENAI_API_KEY` | Yes | OpenAI API key for all LLM calls |
| `TAVILY_API_KEY` | Yes (Claw) | Tavily key for the built-in web-search tool. Get one free at https://app.tavily.com |
Open **http://localhost:8082** — the dashboard comes with 13 built-in pages: Chat, Agents, Memory, Knowledge, Channels, Guardrails, Cron, and more. Add messaging channels directly from the UI.
> 📖 [Full Claw docs](https://docs.praison.ai/docs/concepts/claw) — platform tokens, CLI options, Docker, and YAML agent mode
@@ -382,7 +345,7 @@ praisonai flow
Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
### 8. PraisonAI UI 🤖 (Clean Chat)
+1 -21
View File
@@ -40,7 +40,6 @@ Methods:
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">chat_with_context</a>(message: str, context: 'ContextPack', **kwargs) -> str</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>() -> Optional[bool]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>(value: Optional[bool]) -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">clone_for_channel</a>() -> 'Agent'</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">close</a>() -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">console</a>() -> Optional[Any]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">context_manager</a>() -> Optional[Any]</code>
@@ -172,7 +171,6 @@ Methods:
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">safe_run</a>(**kwargs) -> ToolResult</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_class</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_schema_roundtrip</a>() -> bool</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">__call__</a>(*args, **kwargs) -> Any</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">check_availability</a>() -> tuple[bool, str]</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">injected_params</a>() -> Dict[str, Any]</code>
@@ -185,7 +183,6 @@ Methods:
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_available_tools</a>(context: Optional[Dict[str, Any]] = None) -> List[Union[BaseTool, Callable]]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_base_tools</a>() -> List[BaseTool]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools</a>() -> List[str]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools_with_allowed_filter</a>(context: Optional[Dict[str, Any]] = None) -> List[str]</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, overwrite: bool = False) -> None</code>
* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">unregister</a>(name: str) -> bool</code>
* <code title="class Tools">Tools.<a href="./src/praisonai-agents/praisonaiagents/tools/tools.py">internet_search</a>(*args, **kwargs)</code>
@@ -479,10 +476,8 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">edit_skill</a>(name: str, content: str) -> dict</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_allowed_tools</a>(name: str) -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills</a>() -> List[SkillMetadata]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills_by_state</a>(state: SkillState) -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_instructions</a>(name: str) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skill</a>(name: str) -> Optional[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skills_diagnostics</a>() -> Dict[str, ValidationResult]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_user_invocable_skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">invoke</a>(name: str, raw_args: str = '', session_id: Optional[str] = None, shell_exec: bool = False) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">load_resources</a>(name: str) -> bool</code>
@@ -491,7 +486,6 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skill_names</a>() -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">to_prompt</a>() -> str</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">validate_skill_capabilities</a>(skill_name: str, force_refresh: bool = False) -> ValidationResult</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">write_skill_file</a>(name: str, file_path: str, file_content: str) -> dict</code>
* <code title="class SkillMetadata">SkillMetadata.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">from_properties</a>(props: SkillProperties) -> 'SkillMetadata'</code>
* <code title="class SkillProperties">SkillProperties.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">to_dict</a>() -> dict</code>
@@ -607,13 +601,6 @@ Methods:
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">set_tool_budget</a>(tool_name: str, max_tokens: int, protected: bool = False) -> None</code>
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">truncate_tool_output</a>(tool_name: str, output: str) -> str</code>
# UI
Types:
```python
from praisonaiagents import A2UI
```
# Config
Types:
@@ -768,7 +755,7 @@ Methods:
Types:
```python
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__, arun, run, run_integrated_gateway
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__
```
# CLI
@@ -800,9 +787,6 @@ Methods:
* <code title="cli">praisonai bot email <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot install-daemon <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot linear <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot list <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot purge <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot replay <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot slack <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot start <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
* <code title="cli">praisonai bot telegram <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
@@ -870,10 +854,8 @@ Methods:
* <code title="cli">praisonai doctor env <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor mcp <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor network <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor packaging <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor performance <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor selftest <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor skills <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor tools <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor troubleshoot <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai endpoints list <a href="./src/praisonai/praisonai/cli/commands/endpoints.py">--help</a></code>
@@ -905,7 +887,6 @@ Methods:
* <code title="cli">praisonai gateway send <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway start <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway status <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway stop <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway uninstall <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai github triage <a href="./src/praisonai/praisonai/cli/commands/github.py">--help</a></code>
* <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
@@ -1068,7 +1049,6 @@ Methods:
* <code title="cli">praisonai serve status <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve stop <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui-gateway <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve unified <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai session delete <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
* <code title="cli">praisonai session export <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.48" \
"praisonai>=4.6.37" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.48" \
"praisonai>=4.6.37" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.48" \
"praisonai>=4.6.37" \
"praisonai[ui]" \
"praisonai[crewai]"
-27
View File
@@ -1,27 +0,0 @@
#!/usr/bin/env python3
"""Minimal A2UI example — agent + send_a2ui_messages tool.
Install:
pip install praisonaiagents[a2ui]
Run:
python send_surface.py
For live rendering, use PraisonAIUI with AG-UI (/agui) — see integrate-a2ui-frontend docs.
"""
from praisonaiagents import Agent
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
agent = Agent(
name="ui-assistant",
instructions=(
"When asked for UI, call send_a2ui_messages with A2UI v0.9 JSON "
"(createSurface on surface id 'main')."
),
tools=[send_a2ui_messages],
)
if __name__ == "__main__":
result = agent.start("Show a simple welcome card on surface main")
print(result)
@@ -23,39 +23,12 @@ print(f"[2] Updated to v{managed.agent_version}")
result = agent.start("What is 123 + 456?", stream=True)
# ── 3. Custom tool ──
def _safe_calc(expr: str) -> str:
import ast
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expr):
return "error"
try:
tree = ast.parse(expr, mode="eval")
for node in ast.walk(tree):
if not isinstance(
node,
(
ast.Expression,
ast.BinOp,
ast.UnaryOp,
ast.Constant,
ast.Add,
ast.Sub,
ast.Mult,
ast.Div,
ast.USub,
ast.UAdd,
),
):
return "error"
val = eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
return str(val)
except Exception:
return "error"
def handle_calculator(tool_name, tool_input):
expr = tool_input.get("expression", "0")
val = _safe_calc(str(expr))
try:
val = eval(expr, {"__builtins__": {}})
except Exception:
val = "error"
print(f" [Calculator: {expr} = {val}]")
return str(val)
-40
View File
@@ -1,40 +0,0 @@
# Scientific paper workflow — YAML equivalent of multi_agent_paper.py
# Run with:
# praisonai agents run --file examples/scientific_writing/agents.yaml
framework: praisonai
topic: "Climate change effects on coral reef biodiversity"
roles:
literature_reviewer:
role: "Literature Reviewer"
goal: "Survey recent academic work on {topic} and produce an APA review."
backstory: "Expert in academic literature surveying and citation hygiene."
llm: "gpt-4o-mini"
tasks:
review_literature:
description: "Produce a concise literature review with 5-8 APA citations."
expected_output: "APA-formatted literature review."
methodology_designer:
role: "Methodology Designer"
goal: "Design a reproducible research methodology for {topic}."
backstory: "Specialises in rigorous experimental design."
llm: "gpt-4o-mini"
tasks:
design_methodology:
description: "Write a reproducible methods section in LaTeX."
expected_output: "LaTeX methods section."
scientific_writer:
role: "Scientific Writer"
goal: "Assemble the final LaTeX paper on {topic}."
backstory: >-
Specialised in LaTeX-formatted academic papers, fine-tuned on
scientific literature (CAJAL-4B checkpoint).
llm: "huggingface/Agnuxo/CAJAL-4B-P2PCLAW"
tasks:
write_paper:
description: >-
Combine the literature review and methodology into a full
scientific paper on {topic}. Use LaTeX formatting.
expected_output: "Complete LaTeX paper."
@@ -1,73 +0,0 @@
"""Multi-agent scientific paper workflow (Researcher -> Methodologist -> Writer).
Demonstrates the agent-centric pattern from AGENTS.md: compose specialised
Agents with ``AgentTeam`` + sequential ``Task``s instead of creating custom
Agent subclasses. The writer uses the CAJAL-4B HuggingFace model; the
reviewer and methodologist use general-purpose LLMs.
Usage:
python examples/scientific_writing/multi_agent_paper.py
"""
from praisonaiagents import Agent, AgentTeam, Task
from scientific_writer import ( # noqa: E402 — example imports a sibling file
build_scientific_writer,
format_citation,
format_latex_section,
)
literature_reviewer = Agent(
name="Literature Reviewer",
instructions=(
"You are an expert at surveying academic literature. "
"Produce a concise literature review with 5-8 key citations in APA."
),
tools=[format_citation],
)
methodology_designer = Agent(
name="Methodology Designer",
instructions=(
"You design rigorous research methodologies. "
"Output a clear, reproducible methods section."
),
tools=[format_latex_section],
)
scientific_writer = build_scientific_writer()
def run(topic: str) -> str:
"""Run the 3-agent workflow for a given research ``topic``."""
t1 = Task(
name="review_literature",
description=f"Review literature on: {topic}",
agent=literature_reviewer,
expected_output="APA literature review.",
)
t2 = Task(
name="design_methodology",
description=f"Design research methodology for: {topic}",
agent=methodology_designer,
expected_output="LaTeX methods section.",
)
t3 = Task(
name="write_paper",
description=(
f"Combine the literature review and methodology into a full "
f"scientific paper on: {topic}. Use LaTeX formatting."
),
agent=scientific_writer,
expected_output="Complete LaTeX paper.",
)
team = AgentTeam(
agents=[literature_reviewer, methodology_designer, scientific_writer],
tasks=[t1, t2, t3],
)
return team.start()
if __name__ == "__main__":
result = run("Transformer architectures for protein folding prediction")
print(result)
@@ -1,85 +0,0 @@
"""Scientific Paper Writer — single-agent example using the CAJAL model.
CAJAL (``Agnuxo/CAJAL-4B-P2PCLAW``) is a 4B-parameter HuggingFace model
specialised for LaTeX-formatted academic writing. This example shows how
to use it with a regular ``Agent`` and ``@tool`` functions — no new Agent
subclass is needed. Run 3 ways: Python (this file), CLI, or YAML.
Requirements:
pip install "praisonaiagents[llm]" # for litellm / HuggingFace
export HUGGINGFACE_API_KEY=... # or run a local inference server
Usage:
python examples/scientific_writing/scientific_writer.py
"""
from praisonaiagents import Agent, tool
@tool
def format_latex_section(title: str, content: str) -> str:
"""Wrap prose content in a LaTeX ``\\section{}`` block.
Args:
title: Section heading (e.g. ``"Introduction"``).
content: The body text.
Returns:
LaTeX-formatted section string.
"""
return f"\\section{{{title}}}\n{content}\n"
@tool
def format_citation(authors: str, year: int, title: str, venue: str) -> str:
"""Render a single APA-style citation line.
Args:
authors: ``"Smith, J. & Jones, A."``
year: Publication year.
title: Paper title.
venue: Journal / conference name.
Returns:
One APA-formatted citation string.
"""
return f"{authors} ({year}). {title}. *{venue}*."
SCIENTIFIC_INSTRUCTIONS = """You are a specialised scientific paper writer.
Produce high-quality academic content with:
- Clear structure (Abstract, Introduction, Methodology, Results, Discussion,
Conclusion, References)
- Rigorous methodology and precise language
- Proper APA citations and LaTeX formatting
- Honest limitations and future work sections
Use ``format_latex_section`` to wrap each section in LaTeX, and
``format_citation`` to build the References list.
"""
def build_scientific_writer(
model: str = "huggingface/Agnuxo/CAJAL-4B-P2PCLAW",
) -> Agent:
"""Construct a scientific-writing Agent.
Override ``model`` for any HuggingFace / local / API model — the default
targets the CAJAL-4B checkpoint for LaTeX-style academic output.
"""
return Agent(
name="Scientific Writer",
instructions=SCIENTIFIC_INSTRUCTIONS,
llm=model,
tools=[format_latex_section, format_citation],
)
if __name__ == "__main__":
agent = build_scientific_writer()
paper = agent.start(
"Write a short scientific paper (Abstract + Introduction + Conclusion) "
"on 'Climate change effects on coral reef biodiversity'. "
"Use LaTeX formatting and include at least 3 APA citations."
)
print(paper)
+5 -26
View File
@@ -44,33 +44,12 @@ def main():
def calculate(expression: str) -> str:
"""Calculate a math expression safely."""
import ast
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: Invalid expression"
try:
tree = ast.parse(expression, mode="eval")
for node in ast.walk(tree):
if not isinstance(
node,
(
ast.Expression,
ast.BinOp,
ast.UnaryOp,
ast.Constant,
ast.Add,
ast.Sub,
ast.Mult,
ast.Div,
ast.USub,
ast.UAdd,
),
):
return "Error: Invalid expression"
return str(
eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
)
# Safe eval for basic math
allowed = set("0123456789+-*/.() ")
if all(c in allowed for c in expression):
return str(eval(expression))
return "Error: Invalid expression"
except Exception as e:
return f"Error: {e}"
-222
View File
@@ -1,222 +0,0 @@
# Multi-Channel Bot Deployment Guide
This guide explains how to set up multiple bots on the same platform (e.g., multiple Telegram bots) for different roles using PraisonAI, similar to Hermes-style AI workforces.
## Overview
PraisonAI supports running multiple bots on the same platform, each routing to different agents with specialized knowledge and capabilities. This is useful for:
- **Role-based bots**: `@company_cfo_bot`, `@company_ops_bot`, `@company_content_bot`
- **Department-specific agents**: Each with different knowledge brains and tool access
- **Workflow separation**: Different bots handle different types of requests
## Quick Start with Onboard Wizard
The easiest way to set up multiple channels is using the onboard wizard:
```bash
praisonai onboard
```
The wizard now supports:
1. **Multi-channel setup**: Add multiple bots for the same platform
2. **Role-based naming**: Automatically generates `TELEGRAM_<ROLE>_BOT_TOKEN` environment variables
3. **Agent routing**: Each channel routes to a different agent
### Onboard Wizard Flow
1. **Choose platform**: Start with your first platform (e.g., telegram)
2. **Add more channels**: Wizard asks "Add another bot channel?"
3. **Specify role**: Enter role name (e.g., "cfo", "ops", "content")
4. **Configure tokens**: Each channel gets its own environment variable
5. **Test connections**: Wizard tests each bot separately
## Environment Variable Convention
For multiple bots on the same platform, use this naming convention:
| Channel | Environment Variable | Bot Purpose |
|---------|---------------------|-------------|
| `telegram_cfo` | `TELEGRAM_CFO_BOT_TOKEN` | CFO-related tasks |
| `telegram_ops` | `TELEGRAM_OPS_BOT_TOKEN` | Operations tasks |
| `telegram_content` | `TELEGRAM_CONTENT_BOT_TOKEN` | Content creation |
| `telegram_support` | `TELEGRAM_SUPPORT_BOT_TOKEN` | Customer support |
### Setting Up Tokens
1. **Create bots in @BotFather**:
```text
/newbot
Bot name: Company CFO Bot
Username: @company_cfo_bot
-> Copy token
/newbot
Bot name: Company Ops Bot
Username: @company_ops_bot
-> Copy token
```
2. **Set environment variables**:
```bash
export TELEGRAM_CFO_BOT_TOKEN="123456789:ABC..."
export TELEGRAM_OPS_BOT_TOKEN="987654321:DEF..."
export TELEGRAM_CONTENT_BOT_TOKEN="456789123:GHI..."
```
3. **Or use `~/.praisonai/.env`**:
```env
TELEGRAM_CFO_BOT_TOKEN=123456789:ABC...
TELEGRAM_OPS_BOT_TOKEN=987654321:DEF...
TELEGRAM_CONTENT_BOT_TOKEN=456789123:GHI...
TELEGRAM_ALLOWED_USERS=123456789,987654321
```
## Example Configuration
See `multi-telegram-hermes-workforce.yaml` for a complete example.
### Key Features
1. **Multiple Channels**:
```yaml
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
routes:
default: ops
```
2. **Specialized Agents**:
```yaml
agents:
cfo:
instructions: "You are a CFO agent. Help with financial analysis..."
model: gpt-4o-mini
ops:
instructions: "You are an ops agent. Help with system monitoring..."
model: gpt-4o-mini
```
## Validation with Doctor
Use the doctor command to validate your multi-channel setup:
```bash
praisonai doctor
```
The doctor checks for:
- ✅ **Unique tokens**: Each channel uses a different bot token
- ✅ **Naming convention**: Follows `PLATFORM_ROLE_BOT_TOKEN` pattern
- ✅ **Missing tokens**: All referenced environment variables are set
- ❌ **Duplicate tokens**: Same token used by multiple channels (fails)
## Deployment
### Start Gateway
```bash
# Start the multi-bot gateway
praisonai gateway start --config multi-telegram-hermes-workforce.yaml
# Or install as daemon
praisonai onboard # Wizard installs daemon automatically
praisonai gateway status
```
### Verify Setup
1. **Check health**: `curl http://127.0.0.1:8765/health`
2. **View dashboard**: `praisonai claw` → http://127.0.0.1:8082
3. **Test bots**: Send messages to each bot to verify routing
## Common Patterns
### Basic Multi-Role Setup
```yaml
# Single platform, multiple roles
channels:
telegram_support:
token: ${TELEGRAM_SUPPORT_BOT_TOKEN}
routes:
default: support_agent
telegram_sales:
token: ${TELEGRAM_SALES_BOT_TOKEN}
routes:
default: sales_agent
```
### Cross-Platform Workforce
```yaml
# Multiple platforms, multiple roles
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
discord_ops:
platform: discord
token: ${DISCORD_OPS_BOT_TOKEN}
routes:
default: ops
slack_content:
platform: slack
token: ${SLACK_CONTENT_BOT_TOKEN}
routes:
default: content
```
## Troubleshooting
### Common Issues
1. **"Token is used by multiple channels"**
- Each bot needs its own unique token from @BotFather
- Check environment variables for duplicates
2. **"Channel skipped - no token"**
- Set the environment variable: `export TELEGRAM_CFO_BOT_TOKEN="..."`
- Or update `~/.praisonai/.env`
3. **"Bot not responding"**
- Verify token is correct: `praisonai doctor`
- Check bot is not already running elsewhere
- Ensure allowlist includes your user ID
### Debugging Commands
```bash
# Check configuration
praisonai doctor
# View logs
praisonai gateway logs
# Test connection
curl -H "Authorization: Bearer $GATEWAY_AUTH_TOKEN" \
http://127.0.0.1:8765/info
```
## Migration from Single-Bot Setup
If you currently have a single bot setup, you can add more channels:
1. **Run onboard wizard**: `praisonai onboard`
2. **Add new channels**: Wizard will preserve existing setup
3. **Update configuration**: New channels added to existing `bot.yaml`
4. **Restart gateway**: `praisonai gateway restart`
The wizard preserves backward compatibility with single-bot configurations.
-19
View File
@@ -1,19 +0,0 @@
# A2UI canvas agent — YAML example
#
# Usage:
# pip install praisonaiagents[a2ui]
# praisonai agent run --file a2ui-canvas.yaml --prompt "Show a welcome card"
#
# For live A2UI rendering, use PraisonAIUI (AG-UI / surfaces):
# pip install praisonaiui
# See examples/python/a2ui/send_surface.py and PraisonAIUI example 29.
agent:
name: "ui-assistant"
instructions: |
You are a helpful assistant. When the user asks for UI elements
(buttons, cards, forms), call send_a2ui_messages with valid A2UI v0.9 JSON.
Use surface id "main" unless told otherwise.
llm: gpt-4o-mini
tools:
- send_a2ui_messages
@@ -1,66 +0,0 @@
# Multi-Telegram Channel Hermes Workforce Example
# Generated by `praisonai onboard` with multi-channel support
# Start with: praisonai gateway start --config multi-telegram-hermes-workforce.yaml
gateway:
host: "127.0.0.1"
port: 8765
# Legacy single-agent block for backward compatibility
agent:
name: "assistant"
instructions: "You are a helpful AI assistant. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."
model: "gpt-4o-mini"
# Multi-agent workforce with different specialized roles
agents:
cfo:
instructions: "You are a CFO agent. You are a helpful AI assistant. Use your tools proactively to help with CFO-related tasks like financial analysis, budget planning, and cost optimization."
model: gpt-4o-mini
memory: false
ops:
instructions: "You are an ops agent. You are a helpful AI assistant. Use your tools proactively to help with ops-related tasks like system monitoring, deployment, and infrastructure management."
model: gpt-4o-mini
memory: false
content:
instructions: "You are a content agent. You are a helpful AI assistant. Use your tools proactively to help with content-related tasks like writing, editing, and content strategy."
model: gpt-4o-mini
memory: false
# Multiple Telegram channels - each with unique tokens and routing
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: ops
telegram_content:
platform: telegram
token: ${TELEGRAM_CONTENT_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: content
# Legacy routing block
routing:
default: assistant
@@ -1,10 +1,6 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 real agentic smoke test (high-level BotOS API, no manual mgr).
This is not a pytest test file - it's an integration smoke test.
Uses the user-facing API only — ``Bot``, ``BotOS``, ``InMemoryIdentityResolver`` —
and exercises the same cross-platform-continuity scenario as
``smoke_w1_real.py`` but without touching ``BotSessionManager`` directly.
@@ -17,11 +13,10 @@ wire-up from ``BotOS(identity_resolver=...)`` → ``Bot._identity_resolver``
Run::
# From the repository root:
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_botos_real.py
PYTHONPATH=src/praisonai-agents:src/praisonai python scripts/smoke_w1_botos_real.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
@@ -1,18 +1,13 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 real agentic smoke test (bypasses pytest conftest gating).
This is not a pytest test file - it's an integration smoke test.
Runs::
cd /path/to/PraisonAI
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_real.py
cd src/praisonai
python ../../scripts/smoke_w1_real.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
@@ -1,10 +1,6 @@
#!/usr/bin/env python3
from __future__ import annotations
"""W1 deep-validation smoke test.
This is not a pytest test file - it's an integration smoke test.
Exercises every robustness dimension of W1 with REAL LLM calls:
T1. 3-platform unification (telegram → discord → slack, one user)
@@ -16,12 +12,11 @@ Exercises every robustness dimension of W1 with REAL LLM calls:
Run::
cd /path/to/PraisonAI
PYTHONPATH=src/praisonai-agents:src/praisonai python src/praisonai/tests/integration/smoke_w1_robust.py
cd /Users/praison/worktrees/hermes-parity
PYTHONPATH=src/praisonai-agents:src/praisonai python scripts/smoke_w1_robust.py
"""
# Prevent pytest from treating this as a test file
__test__ = False
from __future__ import annotations
import asyncio
import os
@@ -343,7 +343,6 @@ _LAZY_IMPORTS = {
# UI (optional)
'AGUI': ('praisonaiagents.ui.agui', 'AGUI'),
'A2A': ('praisonaiagents.ui.a2a', 'A2A'),
'A2UI': ('praisonaiagents.ui.a2ui', 'A2UI'),
# Feature configs
'MemoryConfig': ('praisonaiagents.config.feature_configs', 'MemoryConfig'),
@@ -815,6 +814,6 @@ def __dir__():
# - Workflow, Task, Route, Parallel, Loop, Repeat, Pipeline, etc.
# - MCP, FlowDisplay, track_workflow, FastContext, etc.
# - Plan, PlanStep, TodoList, PlanningAgent, ApprovalCallback, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, A2UI, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, etc.
# - All telemetry, display, and utility functions
# ============================================================================
@@ -21,7 +21,6 @@ from .chat_handler import ChatHandlerMixin
from .session_manager import SessionManagerMixin
from .async_safety import AsyncSafeState
from .unified_execution_mixin import UnifiedExecutionMixin
from .sandbox_mixin import SandboxMixin
# Module-level logger for thread safety errors and debugging
logger = get_logger(__name__)
@@ -256,7 +255,7 @@ if TYPE_CHECKING:
# Import structured error from central errors module
from ..errors import BudgetExceededError
class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
# Class-level counter for generating unique display names for nameless agents
_agent_counter = 0
_agent_counter_lock = threading.Lock()
@@ -556,7 +555,6 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
backend: Optional[Any] = None, # External managed agent backend (e.g., ManagedAgentIntegration)
cli_backend: Optional[Union[str, Any]] = None, # CLI backend for delegating turns (e.g., "claude-code")
interrupt_controller: Optional['InterruptController'] = None, # G2: Cooperative cancellation
sandbox: Optional[Union[bool, 'SandboxConfig']] = None, # Sandbox for safe code execution
):
"""Initialize an Agent instance.
@@ -1948,132 +1946,6 @@ Your Goal: {self.goal}
# Telemetry - lazy initialized via property for performance
self.__telemetry = None
self.__telemetry_initialized = False
# Sandbox configuration - initialize SandboxMixin
super().__init__(sandbox=sandbox)
def __deepcopy__(self, memo: dict) -> "Agent":
"""Custom deepcopy that creates fresh threading primitives.
threading.RLock (self.__cache_lock) and threading.Lock (self._cost_lock)
cannot be deep-copied on CPython < 3.13. This hook deep-copies every
other attribute normally and replaces the locks with new instances so
that copy.deepcopy(agent) works in any Python version.
"""
import copy
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
for k, v in self.__dict__.items():
if k in ("_Agent__cache_lock",):
object.__setattr__(result, k, threading.RLock())
elif k == "_cost_lock":
object.__setattr__(result, k, threading.Lock())
else:
object.__setattr__(result, k, copy.deepcopy(v, memo))
return result
def clone_for_channel(self) -> "Agent":
"""Return a fully independent copy of this agent for a gateway channel.
This method safely clones an agent for use in multi-channel scenarios
by avoiding deepcopy issues with tools/handoffs and creating fresh
instances of mutable objects like interrupt_controller.
Returns:
Agent: A new Agent instance with copied configuration but fresh
locks, interrupt controller, and shallow-copied tools.
"""
import copy
from ..tools.base import BaseTool
# Build kwargs for new Agent instance, copying core attributes
clone_kwargs = {
# Core identity
'name': self.name,
'role': self.role,
'goal': self.goal,
'backstory': self.backstory,
'instructions': self.instructions,
# LLM configuration
'llm': self.llm,
'base_url': getattr(self, 'base_url', None),
'api_key': getattr(self, 'api_key', None),
'auth': getattr(self, 'auth', None),
# Shallow copy tools to avoid deepcopy issues with nested objects
'tools': list(self.tools) if self.tools else None,
# Skip handoffs entirely - they shouldn't be shared across channels
# and can contain nested Agent instances that cause RLock issues
'handoffs': None,
# Feature configurations - check for actual stored config objects
'memory': getattr(self, '_memory_config', None),
'knowledge': getattr(self, '_knowledge_config', None),
'planning': getattr(self, '_planning_config', None),
'reflection': getattr(self, '_reflection_config', None),
'guardrails': getattr(self, '_guardrails_config', None),
'web': getattr(self, '_web_config', None),
'context': getattr(self, '_context_config', None),
'autonomy': getattr(self, '_autonomy_config', None),
'output': getattr(self, '_output_config', None),
'execution': getattr(self, '_execution_config', None),
'templates': getattr(self, '_template_config', None),
'caching': getattr(self, '_caching_config', None),
'hooks': getattr(self, '_hooks_config', None),
'skills': getattr(self, '_skills_config', None),
'approval': getattr(self, '_approval_config', None),
'learn': getattr(self, '_learn_config', None),
# Tool configuration
'tool_timeout': getattr(self, '_tool_timeout', None),
'parallel_tool_calls': getattr(self, 'parallel_tool_calls', False),
# CLI backend
'cli_backend': getattr(self, '_cli_backend', None),
# Create fresh interrupt controller to avoid shared state
'interrupt_controller': None, # Let new instance create its own
# Sandbox config
'sandbox': getattr(self, '_sandbox_config', None),
}
# Handle deprecated parameters for backward compatibility
import warnings
# Check for deprecated standalone attributes and emit warnings
deprecated_attrs = [
('allow_code_execution', 'execution'),
('code_execution_mode', 'execution'),
('auto_save', 'memory'),
('rate_limiter', 'execution'),
('allow_delegation', 'handoffs'),
('verification_hooks', 'autonomy')
]
for old_attr, new_param in deprecated_attrs:
if hasattr(self, old_attr):
value = getattr(self, old_attr)
if value is not None and value != False: # Skip None/False defaults
warnings.warn(
f"Deprecated attribute '{old_attr}' found in agent clone. "
f"Use '{new_param}=' parameter instead.",
DeprecationWarning,
stacklevel=2
)
# For basic compatibility, include in clone_kwargs if not already set
if old_attr == 'allow_code_execution' and clone_kwargs['execution'] is None:
from ..config.execution import ExecutionConfig
clone_kwargs['execution'] = ExecutionConfig(code_execution=value)
elif old_attr == 'auto_save' and clone_kwargs['memory'] is None:
from ..config.memory import MemoryConfig
clone_kwargs['memory'] = MemoryConfig(auto_save=value)
# Create new Agent instance
return self.__class__(**{k: v for k, v in clone_kwargs.items() if v is not None})
@property
def _telemetry(self):
@@ -4739,57 +4611,60 @@ Answer:"""
error=f"Agent guardrail validation error: {str(e)}"
)
def _validate_with_guardrail(self, response_text):
"""Validate response with guardrail. Returns (success, result, error)."""
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic.
Args:
response_text: The response to validate
prompt: Original prompt for regeneration if needed
temperature: Temperature for regeneration
tools: Tools for regeneration
Returns:
str: The validated response text or None if validation fails after retries
"""
if not self._guardrail_fn:
return True, response_text, None
return response_text
from ..main import TaskOutput
task_output = TaskOutput(
description="Agent response output",
raw=response_text,
agent=self.name
)
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return True, guardrail_result.result.raw, None
elif guardrail_result.result:
return True, str(guardrail_result.result), None
else:
return True, response_text, None
else:
return False, None, guardrail_result.error
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (sync version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
# Create TaskOutput object
task_output = TaskOutput(
description="Agent response output",
raw=current_response,
agent=self.name
)
if success:
# Process guardrail
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return guardrail_result.result.raw
elif guardrail_result.result:
return str(guardrail_result.result)
else:
return current_response
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Last error: {guardrail_result.error}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {guardrail_result.error}")
# Regenerate response for retry
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {guardrail_result.error}. Please provide an improved response."
response = self._chat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and response.choices:
content = response.choices[0].message.content
@@ -4798,46 +4673,10 @@ Answer:"""
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
async def _aapply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (async version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
if success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
# If we can't regenerate, fail the guardrail
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Agent {self.name} guardrail retry failed: {e}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
# Regenerate response for retry (async version)
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
response = await self._execute_unified_achat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and hasattr(response, 'choices') and response.choices:
content = response.choices[0].message.content
current_response = content.strip() if content else ""
elif isinstance(response, str):
current_response = response.strip()
else:
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
@@ -753,11 +753,11 @@ Your Goal: {self.goal}"""
# Create LLMError with contextual metadata
error = LLMError(
str(e),
model_name=model_name,
agent_id=self.name,
str(e),
model_name=model_name,
agent_id=self.name,
is_retryable=is_retryable,
context={"session_id": session_id},
session_id=session_id
)
# Call error hook if available for error interception
@@ -902,7 +902,7 @@ Your Goal: {self.goal}"""
max_tokens=getattr(self, 'max_tokens', None),
stream=stream,
response_format=response_format,
execute_tool_fn=getattr(self, 'execute_tool_async', None),
execute_tool_fn=getattr(self, 'execute_tool', None),
console=self.console if (self.verbose or stream) else None,
display_fn=self._display_generating if self.verbose else None,
stream_callback=stream_callback,
@@ -935,151 +935,6 @@ Your Goal: {self.goal}"""
logging.error(f"Unified async chat completion failed: {e}")
raise
async def _finalize_unified_achat_response(
self,
response,
original_prompt,
temperature,
tools,
output_json,
output_pydantic,
reasoning_steps,
stream,
messages,
chat_history_length,
start_time,
task_name=None,
task_description=None,
task_id=None,
):
"""Post-process unified achat completion (parity with sync _chat_impl)."""
if not response:
self._truncate_chat_history(chat_history_length)
return None
if hasattr(response, "choices") and response.choices:
msg = response.choices[0].message
content = getattr(msg, "content", None)
response_text = content.strip() if content else ""
if (
reasoning_steps
and getattr(msg, "reasoning_content", None)
):
response_text = msg.reasoning_content.strip()
else:
extracted = self._extract_llm_response_content(response)
response_text = (extracted or "").strip() if isinstance(extracted, str) else str(extracted or "")
if output_json or output_pydantic:
self._append_to_chat_history({"role": "assistant", "content": response_text})
self._persist_message("assistant", response_text)
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
self._truncate_chat_history(chat_history_length)
return None
if not self.self_reflect:
self._append_to_chat_history({"role": "assistant", "content": response_text})
self._persist_message("assistant", response_text)
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
self._truncate_chat_history(chat_history_length)
return None
# Self-reflection (legacy OpenAI path)
reflection_count = 0
while True:
reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
"""
reflection_messages = messages + [
{"role": "assistant", "content": response_text},
{"role": "user", "content": reflection_prompt},
]
try:
if self._openai_client is None:
self._append_to_chat_history({"role": "assistant", "content": response_text})
return await self._atrigger_after_agent_hook(
original_prompt, response_text, start_time,
)
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
messages=reflection_messages,
temperature=temperature,
response_format=_get_display_functions()['ReflectionOutput'],
)
reflection_output = reflection_response.choices[0].message.parsed
if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
break
if reflection_count >= self.max_reflect - 1:
break
regenerate_messages = reflection_messages + [
{
"role": "assistant",
"content": (
f"Self Reflection: {reflection_output.reflection} "
f"Satisfactory?: {reflection_output.satisfactory}"
),
},
{"role": "user", "content": "Now regenerate your response using the reflection you made"},
]
new_response = await self._openai_client.async_client.chat.completions.create(
model=self.llm,
messages=regenerate_messages,
temperature=temperature,
)
response_text = new_response.choices[0].message.content
reflection_count += 1
except Exception as e:
logging.error("Reflection parsing failed.", exc_info=True)
reflection_count += 1
if reflection_count >= self.max_reflect:
break
try:
validated_response = self._apply_guardrail_with_retry(
response_text, original_prompt, temperature, tools,
task_name, task_description, task_id,
)
self._execute_callback_and_display(
original_prompt, validated_response, time.time() - start_time,
task_name, task_description, task_id,
)
return await self._atrigger_after_agent_hook(
original_prompt, validated_response, start_time,
)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for OpenAI client: {e}")
self._truncate_chat_history(chat_history_length)
return None
def _execute_callback_and_display(self, prompt: str, response: str, generation_time: float, task_name=None, task_description=None, task_id=None):
"""Helper method to execute callbacks and display interaction.
@@ -2079,7 +1934,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for custom LLM response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(normalized_content, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(prompt, validated_response, start_time)
@@ -2187,216 +2042,8 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
task_id=task_id,
response_format=response_format
)
# Process response - mirror sync _chat_impl behavior (lines ~1544-1602)
if not response:
# Rollback chat history on response failure
self._truncate_chat_history(chat_history_length)
return None
# Extract response content using the same method as sync
response_text = self._extract_llm_response_content(response)
if isinstance(response_text, str):
response_text = response_text.strip()
# Handle output_json or output_pydantic if specified
if output_json or output_pydantic:
# Add to chat history and return raw response
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
# Apply guardrail validation even for JSON output
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# For regular responses (no self-reflection)
if not self.self_reflect:
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB (non-reflect path)
self._persist_message("assistant", response_text)
if self.verbose:
logging.debug(f"Agent {self.name} final response: {response_text}")
# Return only reasoning content if reasoning_steps is True
if reasoning_steps and hasattr(response, 'choices') and response.choices and hasattr(response.choices[0].message, 'reasoning_content') and response.choices[0].message.reasoning_content:
# Apply guardrail to reasoning content
try:
validated_reasoning = await self._aapply_guardrail_with_retry(response.choices[0].message.reasoning_content, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_reasoning, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_reasoning, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
else:
# Apply guardrail to regular response content
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# If self-reflection is enabled, implement reflection logic
if self.self_reflect:
# Implement async self-reflection similar to sync path
reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
"""
# Add reflection prompt to messages
reflection_messages = messages + [
{"role": "assistant", "content": response_text},
{"role": "user", "content": reflection_prompt}
]
reflection_count = 0
while True:
try:
# Check if OpenAI client is available for self-reflection
if self._openai_client is None:
# For custom LLMs, self-reflection with structured output is not supported
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name}: Self-reflection with structured output is not supported for custom LLM providers. Skipping reflection.", console=self.console)
# Return the original response without reflection
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
messages=reflection_messages,
temperature=temperature,
response_format=_get_display_functions()['ReflectionOutput']
)
reflection_output = reflection_response.choices[0].message.parsed
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name} self reflection (using {self.reflect_llm if self.reflect_llm else self.llm}): reflection='{reflection_output.reflection}' satisfactory='{reflection_output.satisfactory}'", console=self.console)
# Only consider satisfactory after minimum reflections
if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Agent marked the response as satisfactory after meeting minimum reflections", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Check if we've hit max reflections
if reflection_count >= self.max_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Maximum reflection count reached, returning current response", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Regenerate response based on reflection
regenerate_messages = reflection_messages + [
{"role": "assistant", "content": f"Self Reflection: {reflection_output.reflection} Satisfactory?: {reflection_output.satisfactory}"},
{"role": "user", "content": "Now regenerate your response using the reflection you made"}
]
new_response = await self._execute_unified_achat_completion(
messages=regenerate_messages,
temperature=temperature,
tools=formatted_tools,
stream=stream,
reasoning_steps=reasoning_steps,
task_name=task_name,
task_description=task_description,
task_id=task_id,
response_format=response_format
)
if new_response:
new_response_text = self._extract_llm_response_content(new_response)
if isinstance(new_response_text, str):
response_text = new_response_text.strip()
# Update reflection_messages to include the new response for next iteration
reflection_messages = regenerate_messages + [
{"role": "assistant", "content": response_text}
]
reflection_count += 1
except Exception as e:
if self.verbose:
_get_display_functions()['display_error'](f"Error in parsing self-reflection json {e}. Retrying", console=self.console)
logging.error("Reflection parsing failed.", exc_info=True)
reflection_count += 1
if reflection_count >= self.max_reflect:
# Return original response after max reflection attempts
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as guard_e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection error: {guard_e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
continue
# This should never be reached due to the returns above
# But adding as safety fallback
logging.warning(f"Agent {self.name}: Unexpected code path reached in unified dispatch")
return None
# Continue to handle structured outputs and other processing
# Don't return immediately - fall through to existing logic
else:
# LEGACY: Check if OpenAI client is available
if self._openai_client is None:
@@ -2533,7 +2180,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for OpenAI client response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
@@ -164,8 +164,7 @@ Follow best practices and coding standards."""
messages=messages,
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
code = response.choices[0].message.content
self._log("Code generated successfully", "green")
return code
@@ -205,8 +204,7 @@ Follow best practices and coding standards."""
messages=messages,
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
code = response.choices[0].message.content
self._log("Code generated successfully", "green")
return code
@@ -367,8 +365,6 @@ Provide clear, actionable feedback."""
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
review = response.choices[0].message.content
self._log("Review completed", "green")
return review
@@ -409,8 +405,6 @@ Include:
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
explanation = response.choices[0].message.content
self._log("Explanation completed", "green")
return explanation
@@ -457,8 +451,6 @@ Only output the refactored code."""
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
refactored = response.choices[0].message.content
self._log("Refactoring completed", "green")
return refactored
@@ -503,8 +495,6 @@ Ensure the fix:
**{k: v for k, v in kwargs.items() if k != "model"}
)
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
fixed = response.choices[0].message.content
self._log("Fix completed", "green")
return fixed
@@ -403,33 +403,9 @@ class MemoryMixin:
self._session_store.add_user_message(self._session_id, content)
elif role == "assistant":
self._session_store.add_assistant_message(self._session_id, content)
self._persist_session_stats()
except Exception as e:
logging.warning(f"Failed to persist message to session store: {e}")
def _persist_session_stats(self):
"""Flush agent cost/token stats into session JSON metadata."""
store = getattr(self, "_session_store", None)
session_id = getattr(self, "_session_id", None)
if store is None or session_id is None:
return
if not hasattr(store, "update_session_metadata"):
return
try:
summary = getattr(self, "cost_summary", {}) or {}
llm = getattr(self, "llm", None)
model = str(llm) if llm is not None else None
store.update_session_metadata(
session_id,
model=model,
total_tokens=summary.get("tokens_in", 0) + summary.get("tokens_out", 0),
cost=summary.get("cost", 0),
source=getattr(self, "_session_source", "chat"),
agent_id=getattr(self, "agent_id", None) or getattr(self, "_agent_id", None),
)
except Exception as e:
logging.debug(f"Session stats persist skipped: {e}")
def session_id(self) -> Optional[str]:
"""Get the current session ID."""
return self._session_id
@@ -1,231 +0,0 @@
"""
Sandbox execution mixin for Agent.
Provides safe code execution capabilities to agents via the sandbox framework.
"""
import logging
from typing import Optional, Dict, Any, Union
from ..sandbox import (
SandboxConfig,
SandboxManager,
SandboxResult,
check_code_safety,
format_warnings,
)
logger = logging.getLogger(__name__)
class SandboxMixin:
"""Mixin that adds sandbox execution capabilities to Agent.
Provides safe code execution in isolated environments with security pre-checks.
"""
def __init__(self, *args, **kwargs):
# Extract sandbox config from kwargs
self.sandbox_config: Optional[SandboxConfig] = kwargs.pop('sandbox', None)
# Convert boolean True to default config
if self.sandbox_config is True:
self.sandbox_config = SandboxConfig.subprocess()
super().__init__(*args, **kwargs)
self._sandbox_manager: Optional[SandboxManager] = None
@property
def has_sandbox(self) -> bool:
"""Check if sandbox is configured for this agent."""
return self.sandbox_config is not None
def get_sandbox_manager(self) -> Optional[SandboxManager]:
"""Get or create sandbox manager."""
if not self.has_sandbox:
return None
if self._sandbox_manager is None:
self._sandbox_manager = SandboxManager(self.sandbox_config)
return self._sandbox_manager
async def execute_code(
self,
code: str,
language: str = "python",
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Execute code safely in configured sandbox.
Args:
code: Code to execute
language: Programming language (python, bash, etc.)
check_security: Whether to run security pre-checks
**kwargs: Additional arguments passed to sandbox.execute()
Returns:
SandboxResult with execution details
Raises:
RuntimeError: If no sandbox is configured
"""
if not self.has_sandbox:
raise RuntimeError(
"No sandbox configured. Set sandbox=True or provide SandboxConfig to Agent()"
)
# Security pre-check if enabled
if check_security:
warnings = check_code_safety(code, language)
if warnings:
warning_text = format_warnings(warnings)
if self.verbose:
logger.warning(f"Security warnings for code execution:\n{warning_text}")
# Store warnings in result metadata
kwargs.setdefault('metadata', {})['security_warnings'] = warnings
manager = self.get_sandbox_manager()
return await manager.run_code(code, language=language, **kwargs)
def execute_code_sync(
self,
code: str,
language: str = "python",
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Synchronous wrapper for execute_code.
Args:
code: Code to execute
language: Programming language
check_security: Whether to run security pre-checks
**kwargs: Additional arguments
Returns:
SandboxResult with execution details
"""
from ..approval.utils import run_coroutine_safely
return run_coroutine_safely(self.execute_code(code, language, check_security, **kwargs))
async def run_shell_command(
self,
command: Union[str, list],
check_security: bool = True,
**kwargs
) -> SandboxResult:
"""Run a shell command in the sandbox.
Args:
command: Shell command to run
check_security: Whether to run security pre-checks
**kwargs: Additional arguments
Returns:
SandboxResult with execution details
"""
if not self.has_sandbox:
raise RuntimeError("No sandbox configured")
# Convert command to string for security checking
if isinstance(command, list):
command_str = " ".join(command)
else:
command_str = command
# Security pre-check
if check_security:
warnings = check_code_safety(command_str, "bash")
if warnings:
warning_text = format_warnings(warnings)
if self.verbose:
logger.warning(f"Security warnings for command:\n{warning_text}")
async with self.get_sandbox_manager() as sandbox:
return await sandbox.run_command(command, **kwargs)
def get_sandbox_status(self) -> Dict[str, Any]:
"""Get sandbox status information.
Returns:
Dictionary with sandbox status details
"""
if not self.has_sandbox:
return {"configured": False}
status = {
"configured": True,
"config": self.sandbox_config.to_dict() if self.sandbox_config else None,
}
manager = self.get_sandbox_manager()
if manager:
# Get available sandbox types
available_types = manager.get_available_types()
status["available_types"] = available_types
status["current_type"] = self.sandbox_config.sandbox_type
return status
async def sandbox_cleanup(self):
"""Clean up sandbox resources."""
if self._sandbox_manager:
# The context manager handles cleanup, but we can force it here
try:
await self._sandbox_manager.__aexit__(None, None, None)
except Exception as e:
logger.warning(f"Error during sandbox cleanup: {e}")
finally:
self._sandbox_manager = None
def _get_code_execution_tools(self):
"""Get sandbox-based code execution tools for this agent.
Returns tools that use the configured sandbox for safe execution.
Returns:
List of tool functions
"""
if not self.has_sandbox:
return []
from ..tools import tool
@tool
def execute_python_code(code: str) -> str:
"""Execute Python code safely in sandbox.
Args:
code: Python code to execute
Returns:
Execution output
"""
result = self.execute_code_sync(code, language="python")
if result.success:
return result.stdout or result.output
else:
return f"Error: {result.error or result.stderr}"
@tool
def execute_shell_command(command: str) -> str:
"""Execute shell command safely in sandbox.
Args:
command: Shell command to execute
Returns:
Command output
"""
from ..approval.utils import run_coroutine_safely
result = run_coroutine_safely(self.run_shell_command(command))
if result.success:
return result.stdout or result.output
else:
return f"Error: {result.error or result.stderr}"
return [execute_python_code, execute_shell_command]
@@ -26,21 +26,6 @@ if TYPE_CHECKING:
class ToolExecutionMixin:
"""Mixin providing toolexecution methods for the Agent class."""
def _get_existing_stream_emitter(self):
"""Return an already-initialized stream emitter without creating one."""
emitter = getattr(self, "_stream_emitter", None)
if emitter is not None:
return emitter
# Support name-mangled private attributes across class renames/inheritance.
for cls in type(self).mro():
mangled = f"_{cls.__name__}__stream_emitter"
if hasattr(self, mangled):
emitter = getattr(self, mangled, None)
if emitter is not None:
return emitter
return None
def _resolve_tool_names(self, tool_names):
"""Resolve tool names to actual tool instances from registry.
@@ -175,9 +160,8 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_START to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
_stream_emitter = self._get_existing_stream_emitter()
if _stream_emitter is not None and _stream_emitter.has_callbacks:
_stream_emitter.emit(StreamEvent(
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_START,
timestamp=_tool_start_perf,
tool_call={
@@ -291,10 +275,10 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_RESULT to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
if _stream_emitter is not None and _stream_emitter.has_callbacks:
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
# Truncate result for stream event (keep it reasonable for UI display)
result_summary = str(result)[:500] if result else None
_stream_emitter.emit(StreamEvent(
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_RESULT,
timestamp=_time.perf_counter(),
tool_call={
@@ -306,18 +290,6 @@ class ToolExecutionMixin:
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
_stream_emitter.emit(StreamEvent(
type=StreamEventType.TOOL_CALL_END,
timestamp=_time.perf_counter(),
tool_call={
"name": function_name,
"arguments": arguments,
"result": result_summary,
"id": tool_call_id,
},
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
# Trigger AFTER_TOOL hook
from ..hooks import HookEvent, AfterToolInput
@@ -954,3 +926,4 @@ class ToolExecutionMixin:
def pending_approval_count(self) -> int:
"""Number of approval requests still waiting."""
return len(self._pending_approvals)
@@ -15,9 +15,9 @@ from enum import Enum
# Import token tracking
try:
from ..telemetry.token_collector import get_token_collector
from ..telemetry.token_collector import _token_collector
except ImportError:
get_token_collector = None
_token_collector = None
# Import async utility for hot-path usage
try:
@@ -126,24 +126,7 @@ class _AgentServerRegistry:
thread = threading.Thread(target=run_server, daemon=True)
thread.start()
# Check for configurable timeout via environment variable
try:
timeout = float(os.environ.get("PRAISONAI_SERVER_READY_TIMEOUT", "5.0"))
except ValueError:
logger.warning("Invalid PRAISONAI_SERVER_READY_TIMEOUT value. Using default 5.0s.")
timeout = 5.0
became_ready = ready_event.wait(timeout=timeout)
if not became_ready:
logger.warning(
"Agent server on port %s did not become ready within %.1fs. "
"Proceeding, but some features may not work correctly. "
"Check server logs for startup errors.",
port,
timeout,
)
ready_event.wait(timeout=5.0) # Deterministic wait instead of sleep(0.5)
return True
@@ -1609,25 +1592,25 @@ class AgentTeam:
original_on_task_start = self.on_task_start
original_on_task_complete = self.on_task_complete
def composed_on_task_start(task, task_id):
def composed_on_task_start(task):
try:
verbose_task_start_callback(task, task_id)
verbose_task_start_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task start callback: {e}")
if original_on_task_start:
try:
original_on_task_start(task, task_id)
original_on_task_start(task)
except Exception as e:
logging.debug(f"Error in original task start callback: {e}")
def composed_on_task_complete(task, task_output):
def composed_on_task_complete(task):
try:
verbose_task_complete_callback(task, task_output)
verbose_task_complete_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task complete callback: {e}")
if original_on_task_complete:
try:
original_on_task_complete(task, task_output)
original_on_task_complete(task)
except Exception as e:
logging.debug(f"Error in original task complete callback: {e}")
@@ -1826,19 +1809,18 @@ class AgentTeam:
def get_token_usage_summary(self) -> Dict[str, Any]:
"""Get a summary of token usage across all agents and tasks."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
return get_token_collector().get_session_summary()
return _token_collector.get_session_summary()
def get_detailed_token_report(self) -> Dict[str, Any]:
"""Get a detailed token usage report."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
collector = get_token_collector()
summary = collector.get_session_summary()
recent = collector.get_recent_interactions(limit=20)
summary = _token_collector.get_session_summary()
recent = _token_collector.get_recent_interactions(limit=20)
# Calculate cost estimates (example rates)
cost_per_1k_input = 0.0005 # $0.0005 per 1K input tokens
@@ -1862,11 +1844,11 @@ class AgentTeam:
def display_token_usage(self):
"""Display token usage in a formatted table."""
if not get_token_collector:
if not _token_collector:
print("Token tracking not available")
return
summary = get_token_collector().get_session_summary()
summary = _token_collector.get_session_summary()
print("\n" + "="*50)
print("TOKEN USAGE SUMMARY")
@@ -1,257 +0,0 @@
"""
Tool whitelist filter module with diagnostics.
This module provides a canonical implementation of the ALLOWED_TOOLS filter
that prevents tool name collisions in multi-environment agent systems by
whitelisting only specified tools.
The filter is designed to solve the "tool shadowing" problem where multiple
modules register tools with overlapping or ambiguous names, causing agents
to invoke the wrong implementation.
Environment Variables:
ALLOWED_TOOLS: Primary variable (comma-separated tool names)
HERMES_ONLY_TOOLS: Backward compatibility alias for ALLOWED_TOOLS
Usage:
from praisonaiagents.allowed_tools_filter import AllowedToolsFilter
# Create filter from environment variable
filter = AllowedToolsFilter()
# Filter tools
available_tools = {"search", "send_message", "extract_pdf"}
filtered_tools = filter.filter_tools(available_tools)
# Get diagnostics
filter.log_diagnostics()
"""
import os
import logging
from typing import Any, Dict, List, Optional, Set, Union
logger = logging.getLogger(__name__)
class AllowedToolsFilter:
"""
Canonical tool whitelist filter with diagnostics and backward compatibility.
This filter implements the semantics for tool filtering:
- ALLOWED_TOOLS unset (or HERMES_ONLY_TOOLS): all tools visible (with warning about collisions)
- ALLOWED_TOOLS empty string: error - must specify tools or unset
- ALLOWED_TOOLS with values: only whitelisted tools visible
- Unknown tools in list: warn and strip in dev, strict fail in CI
Backward Compatibility:
- Supports both ALLOWED_TOOLS and HERMES_ONLY_TOOLS
- ALLOWED_TOOLS takes precedence if both are set
"""
def __init__(self, env_var_name: str = "ALLOWED_TOOLS"):
"""
Initialize the filter with backward compatibility support.
Args:
env_var_name: Primary environment variable name (default: "ALLOWED_TOOLS")
"""
# Support both new (ALLOWED_TOOLS) and legacy (HERMES_ONLY_TOOLS) naming
self.primary_var = env_var_name
self.legacy_var = "HERMES_ONLY_TOOLS"
# ALLOWED_TOOLS takes precedence over HERMES_ONLY_TOOLS for backward compatibility
self.env_value = os.environ.get(self.primary_var) or os.environ.get(self.legacy_var)
self.env_var_name = self.primary_var if os.environ.get(self.primary_var) else self.legacy_var
self.is_ci = os.environ.get("CI", "").lower() in ("true", "1", "yes")
self._whitelist: Optional[Set[str]] = self._parse_whitelist()
self._diagnostics: Dict[str, Any] = {}
def _parse_whitelist(self) -> Optional[Set[str]]:
"""
Parse the environment variable value into a whitelist set.
Returns:
None if unset, Set of tool names if set, raises ValueError for empty string
"""
if self.env_value is None:
return None
# Empty string is an error - must be explicit
if self.env_value.strip() == "":
raise ValueError(
f"{self.env_var_name} cannot be empty. Either unset it for all tools "
"or provide a comma-separated list of tool names."
)
# Parse comma-separated values, strip whitespace
tools = {name.strip() for name in self.env_value.split(",") if name.strip()}
if not tools:
raise ValueError(
f"{self.env_var_name} contains no valid tool names. "
"Provide comma-separated tool names or unset the variable."
)
return tools
def filter_tools(self, available_tools: Union[Set[str], List[str], Dict[str, Any]]) -> Set[str]:
"""
Filter tools based on ALLOWED_TOOLS whitelist.
Args:
available_tools: Available tools as set, list, or dict with tool names as keys
Returns:
Filtered set of tool names
Raises:
ValueError: If unknown tools in CI mode or configuration errors
"""
# Convert input to set of tool names
if isinstance(available_tools, dict):
tool_names = set(available_tools.keys())
else:
tool_names = set(available_tools)
# Store for diagnostics
self._diagnostics["registered_before_filter"] = sorted(tool_names)
# If no whitelist, return all tools (with warning)
if self._whitelist is None:
logger.warning(
"Tool whitelist is unset. All %d tools are visible. "
"Consider using ALLOWED_TOOLS to prevent tool name collisions.",
len(tool_names)
)
self._diagnostics["registered_after_filter"] = sorted(tool_names)
self._diagnostics["dropped_tools"] = []
self._diagnostics["unknown_tools"] = []
return tool_names
# Find intersection and unknown tools
available_whitelisted = tool_names.intersection(self._whitelist)
unknown_tools = self._whitelist.difference(tool_names)
dropped_tools = tool_names.difference(self._whitelist)
# Store for diagnostics
self._diagnostics["registered_after_filter"] = sorted(available_whitelisted)
self._diagnostics["dropped_tools"] = sorted(dropped_tools)
self._diagnostics["unknown_tools"] = sorted(unknown_tools)
# Handle unknown tools based on environment
if unknown_tools:
unknown_list = sorted(unknown_tools)
if self.is_ci:
raise ValueError(
f"Unknown tools in {self.env_var_name}: {unknown_list}. "
"All specified tools must be available in CI mode."
)
else:
logger.warning(
"Unknown tools in %s will be ignored: %s",
self.env_var_name,
unknown_list
)
if dropped_tools:
logger.info(
"Dropped %d tools not in %s whitelist: %s",
len(dropped_tools),
self.env_var_name,
sorted(dropped_tools)
)
logger.info(
"%s filtered tools: %d available, %d after filter",
self.env_var_name,
len(tool_names),
len(available_whitelisted)
)
return available_whitelisted
def log_diagnostics(self) -> None:
"""Log startup diagnostics section as specified in the issue."""
if not self._diagnostics:
logger.warning("No diagnostics available. Call filter_tools() first.")
return
logger.info("=" * 50)
logger.info("ALLOWED_TOOLS FILTER DIAGNOSTICS")
logger.info("=" * 50)
logger.info("%s=%s", self.env_var_name, self.env_value or "<unset>")
logger.info("RegisteredBeforeFilter=%s", self._diagnostics.get("registered_before_filter", []))
logger.info("RegisteredAfterFilter=%s", self._diagnostics.get("registered_after_filter", []))
logger.info("DroppedTools=%s", self._diagnostics.get("dropped_tools", []))
logger.info("UnknownTools=%s", self._diagnostics.get("unknown_tools", []))
logger.info("IsCI=%s", self.is_ci)
logger.info("=" * 50)
def get_diagnostics(self) -> Dict[str, Any]:
"""
Get diagnostics data as a dictionary.
Returns:
Dictionary with diagnostics information
"""
return {
"env_var_name": self.env_var_name,
"env_value": self.env_value,
"is_ci": self.is_ci,
"whitelist": list(self._whitelist) if self._whitelist else None,
**self._diagnostics
}
def is_enabled(self) -> bool:
"""
Check if filtering is enabled.
Returns:
True if tool whitelist is set and filtering is active
"""
return self._whitelist is not None
def get_whitelist(self) -> Optional[Set[str]]:
"""
Get the current whitelist.
Returns:
Set of whitelisted tool names, or None if not set
"""
return self._whitelist.copy() if self._whitelist else None
def filter_tools_with_allowed_tools(
available_tools: Union[Set[str], List[str], Dict[str, Any]],
env_var_name: str = "ALLOWED_TOOLS",
log_diagnostics: bool = True
) -> Set[str]:
"""
Convenience function to filter tools with ALLOWED_TOOLS.
Args:
available_tools: Available tools to filter
env_var_name: Environment variable name (default: "ALLOWED_TOOLS")
log_diagnostics: Whether to log diagnostics (default: True)
Returns:
Filtered set of tool names
"""
filter_instance = AllowedToolsFilter(env_var_name)
filtered = filter_instance.filter_tools(available_tools)
if log_diagnostics:
filter_instance.log_diagnostics()
return filtered
# Backward compatibility aliases
filter_tools_with_hermes = filter_tools_with_allowed_tools
hermes_filter = filter_tools_with_allowed_tools
apply_hermes_filter = filter_tools_with_allowed_tools
# Legacy class alias for backward compatibility
HermesToolFilter = AllowedToolsFilter
@@ -42,7 +42,6 @@ class MemoryBackend(str, Enum):
FILE = "file"
SQLITE = "sqlite"
REDIS = "redis"
VALKEY = "valkey"
POSTGRES = "postgres"
MEM0 = "mem0"
MONGODB = "mongodb"
@@ -7,7 +7,7 @@ Inspired by Agno's SessionContextStore pattern.
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional
from datetime import datetime, timezone
from datetime import datetime
import json
import logging
from praisonaiagents._logging import get_logger
@@ -97,8 +97,8 @@ class SessionContextTracker:
self._state = SessionState(
session_id=self.session_id,
created_at=datetime.now(timezone.utc).isoformat(),
updated_at=datetime.now(timezone.utc).isoformat(),
created_at=datetime.utcnow().isoformat() + "Z",
updated_at=datetime.utcnow().isoformat() + "Z",
)
def _generate_session_id(self) -> str:
@@ -163,7 +163,7 @@ class SessionContextTracker:
def _mark_updated(self) -> None:
"""Update the updated_at timestamp."""
self._state.updated_at = datetime.now(timezone.utc).isoformat()
self._state.updated_at = datetime.utcnow().isoformat() + "Z"
self._state.turn_count += 1
def to_context_string(self) -> str:
@@ -257,7 +257,7 @@ Current messages take precedence if there's any conflict with this summary.
self._state = SessionState(
session_id=self.session_id,
created_at=self._state.created_at,
updated_at=datetime.now(timezone.utc).isoformat(),
updated_at=datetime.utcnow().isoformat() + "Z",
)
def to_dict(self) -> Dict[str, Any]:
@@ -12,7 +12,7 @@ import os
import logging
from praisonaiagents._logging import get_logger
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from datetime import datetime
from typing import Any, Dict, List, Optional
logger = get_logger(__name__)
@@ -39,7 +39,7 @@ class GradeResult:
input_text: str = ""
output: str = ""
expected_output: Optional[str] = None
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
@@ -55,7 +55,7 @@ class GradeResult:
input_text=data.get("input_text", ""),
output=data.get("output", ""),
expected_output=data.get("expected_output"),
timestamp=data.get("timestamp", datetime.now(timezone.utc).isoformat()),
timestamp=data.get("timestamp", datetime.utcnow().isoformat()),
)
class BaseLLMGrader:
@@ -1,13 +0,0 @@
"""Hooks query protocol for UI/API consumers."""
from __future__ import annotations
from typing import Dict, List, Protocol, runtime_checkable
@runtime_checkable
class HooksQueryProtocol(Protocol):
"""Read path for registered hooks."""
def list_hooks_for_api(self) -> List[Dict]:
...
@@ -439,13 +439,3 @@ def has_hook(event: Union[str, HookEvent]) -> bool:
)
return get_default_registry().has_hooks(event)
def list_hooks_for_api() -> List[Dict]:
"""Flat hook list suitable for REST/UI consumers."""
registry = get_default_registry()
flat: List[Dict] = []
for event, hooks in registry.list_hooks().items():
for h in hooks:
flat.append({"event": event, "source": "sdk", **h})
return flat
@@ -68,14 +68,6 @@ class HookEvent(str, Enum):
SCHEDULE_REMOVE = "schedule_remove"
SCHEDULE_TRIGGER = "schedule_trigger"
# Kanban task lifecycle (wrapper dispatcher + tools emit these)
KANBAN_TASK_CREATED = "kanban_task_created"
KANBAN_TASK_CLAIMED = "kanban_task_claimed"
KANBAN_TASK_MOVED = "kanban_task_moved"
KANBAN_TASK_DONE = "kanban_task_done"
KANBAN_TASK_BLOCKED = "kanban_task_blocked"
KANBAN_TASK_FAILED = "kanban_task_failed"
# Claude Code parity events
USER_PROMPT_SUBMIT = "user_prompt_submit" # When user submits a prompt
NOTIFICATION = "notification" # When notification is sent
@@ -109,34 +101,6 @@ class HookInput:
}
@dataclass
class KanbanHookInput(HookInput):
"""Hook input for kanban task lifecycle events.
Used by wrapper dispatcher and tools when emitting kanban events.
Observability adapters can subscribe to these via the hook registry.
"""
task_id: str = ""
board: str = "default"
status: str = ""
assignee: Optional[str] = field(default=None)
from_status: Optional[str] = field(default=None)
to_status: Optional[str] = field(default=None)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
result = super().to_dict()
result.update({
"task_id": self.task_id,
"board": self.board,
"status": self.status,
"assignee": self.assignee,
"from_status": self.from_status,
"to_status": self.to_status,
})
return result
@dataclass
class HookOutput:
"""Base hook output - common fields for all events."""
@@ -1,23 +0,0 @@
"""
Kanban protocols and types for PraisonAI Agents.
This module provides the protocol contracts for kanban functionality,
allowing the wrapper (praisonai) and PraisonAIUI to share a stable
interface without coupling the core to SQLite or other heavy implementations.
"""
from praisonaiagents.kanban.protocols import (
KanbanStoreProtocol,
KanbanTaskProtocol,
KanbanCommentingProtocol,
KanbanLinkingProtocol,
VALID_KANBAN_STATUSES,
)
__all__ = [
"KanbanStoreProtocol",
"KanbanTaskProtocol",
"KanbanCommentingProtocol",
"KanbanLinkingProtocol",
"VALID_KANBAN_STATUSES",
]
@@ -1,221 +0,0 @@
"""
Kanban protocols for PraisonAI Agents.
Defines the protocol contracts that the wrapper (praisonai) and PraisonAIUI
can use to implement kanban functionality with a stable interface.
This follows AGENTS.md §3.2 Protocol-First Design:
- Protocols define WHAT (interface contract)
- Implementations define HOW (concrete behavior)
- Core SDK has protocols only, wrapper has heavy implementations
"""
from typing import Protocol, TypedDict, runtime_checkable
# Valid kanban statuses matching PraisonAIUI columns + archived
VALID_KANBAN_STATUSES: frozenset[str] = frozenset([
"triage",
"todo",
"ready",
"running",
"blocked",
"review",
"done",
"archived"
])
class KanbanTaskProtocol(TypedDict, total=False):
"""Typed dict shape for kanban task fields.
Defines the expected structure of task objects returned by
KanbanStoreProtocol implementations.
"""
id: str
title: str
body: str
status: str
assignee: str | None
priority: str | None
tenant: str | None
board: str
created_at: float
updated_at: float
@runtime_checkable
class KanbanStoreProtocol(Protocol):
"""Protocol contract for kanban store implementations.
This protocol defines the core interface that PraisonAIUI expects
for injected kanban stores. Wrapper implementations must
implement all methods to be compatible.
Duck typing contract matches InjectedKanbanStore in PraisonAIUI.
"""
def get_board(
self,
*,
board: str = "default",
tenant: str | None = None,
include_archived: bool = False,
) -> dict:
"""Get board data with tasks grouped by status.
Args:
board: Board name (default: "default")
tenant: Optional tenant filter
include_archived: Whether to include archived tasks
Returns:
Board data with tasks organized by status columns
"""
...
def get_task(self, task_id: str) -> dict | None:
"""Get a single task by ID.
Args:
task_id: Task identifier
Returns:
Task data or None if not found
"""
...
def create_task(self, data: dict) -> dict:
"""Create a new task.
Args:
data: Task creation data (title, body, status, etc.)
Returns:
Created task data with assigned ID
"""
...
def update_task(self, task_id: str, data: dict) -> dict | None:
"""Update an existing task.
Args:
task_id: Task identifier
data: Fields to update
Returns:
Updated task data or None if not found
"""
...
def move_task(self, task_id: str, status: str) -> dict | None:
"""Move a task to a different status.
Args:
task_id: Task identifier
status: New status (must be in VALID_KANBAN_STATUSES)
Returns:
Updated task data or None if not found
"""
...
def bulk_update(self, task_ids: list[str], status: str) -> dict:
"""Update multiple tasks to the same status.
Args:
task_ids: List of task identifiers
status: New status for all tasks
Returns:
Results summary with success/failure counts
"""
...
def delete_task(self, task_id: str) -> bool:
"""Delete a task.
Args:
task_id: Task identifier
Returns:
True if deleted, False if not found
"""
...
def list_events(self, since: float = 0.0, board: str = "default") -> list[dict]:
"""List kanban events since a timestamp.
Args:
since: Unix timestamp to filter events from
board: Board name to filter events
Returns:
List of event data
"""
...
def health(self) -> dict:
"""Check store health status.
Returns:
Health status information
"""
...
@runtime_checkable
class KanbanCommentingProtocol(Protocol):
"""Extension protocol for kanban task commenting functionality.
This protocol is implemented separately from KanbanStoreProtocol to allow
stores to optionally support commenting without breaking isinstance checks
on the core protocol.
"""
def add_comment(self, task_id: str, text: str, author: str | None = None) -> dict | None:
"""Add a comment to a task.
Args:
task_id: Task identifier
text: Comment text
author: Optional comment author
Returns:
Comment data or None if task not found
"""
...
@runtime_checkable
class KanbanLinkingProtocol(Protocol):
"""Extension protocol for kanban task linking functionality.
This protocol is implemented separately from KanbanStoreProtocol to allow
stores to optionally support task relationships without breaking isinstance
checks on the core protocol.
"""
def link_tasks(self, parent_id: str, child_id: str) -> bool:
"""Link two tasks in parent-child relationship.
Args:
parent_id: Parent task identifier
child_id: Child task identifier
Returns:
True if linked successfully, False otherwise
"""
...
def unlink_tasks(self, parent_id: str, child_id: str) -> bool:
"""Unlink parent-child task relationship.
Args:
parent_id: Parent task identifier
child_id: Child task identifier
Returns:
True if unlinked successfully, False otherwise
"""
...
@@ -174,14 +174,7 @@ class ChromaKnowledgeAdapter:
# Add where filter if any conditions specified
if where_filter:
# ChromaDB validation expects exactly one condition in the where clause
# When multiple conditions exist, use $and operator to combine them
if len(where_filter) == 1:
query_kwargs["where"] = where_filter
else:
# Convert multiple conditions to ChromaDB's $and format
and_conditions = [{k: v} for k, v in where_filter.items()]
query_kwargs["where"] = {"$and": and_conditions}
query_kwargs["where"] = where_filter
response = self.collection.query(**query_kwargs)
@@ -204,10 +197,6 @@ class ChromaKnowledgeAdapter:
return SearchResult(results=items)
except Exception as e:
# Log ChromaDB query errors for better debugging
import logging
logger = logging.getLogger(__name__)
logger.warning(f"ChromaDB search failed: {e}")
return SearchResult(results=[])
def add(self, content: Any, *, user_id: Optional[str] = None, agent_id: Optional[str] = None,
@@ -73,10 +73,10 @@ def _get_live():
# Import token tracking
try:
from ..telemetry.token_collector import TokenMetrics, get_token_collector
from ..telemetry.token_collector import TokenMetrics, _token_collector
except ImportError:
TokenMetrics = None
get_token_collector = None
_token_collector = None
# Logging is already configured in _logging.py via __init__.py
@@ -4519,7 +4519,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
def _track_token_usage(self, response: Dict[str, Any], model: str) -> Optional[TokenMetrics]:
"""Extract and track token usage from LLM response."""
if not TokenMetrics or not get_token_collector:
if not TokenMetrics or not _token_collector:
return None
# Note: metrics check moved to call sites for performance
@@ -4551,7 +4551,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Track in global collector
# Extract provider from model string (e.g., 'openai' from 'openai/gpt-4o-mini')
provider = model.split('/')[0] if '/' in model else 'litellm'
get_token_collector().track_tokens(
_token_collector.track_tokens(
model=model,
agent=self.current_agent_name,
metrics=metrics,
@@ -907,8 +907,6 @@ class OpenAIClient:
final_response = self._responses_to_chat_completion(raw)
# Display the response text if display_fn provided
if not final_response.choices or final_response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
response_text = final_response.choices[0].message.content or ""
if display_fn and response_text:
from rich.markdown import Markdown
@@ -1169,8 +1167,6 @@ class OpenAIClient:
final_response = self._responses_to_chat_completion(raw)
# Display the response text if display_fn provided
if not final_response.choices or final_response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
response_text = final_response.choices[0].message.content or ""
if display_fn and response_text:
from rich.markdown import Markdown
@@ -1642,10 +1638,8 @@ class OpenAIClient:
)
# Check for tool calls
if not final_response.choices or final_response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
tool_calls = getattr(final_response.choices[0].message, 'tool_calls', None)
# Emit llm_content for intermediate narrative display
# (gpt-4.1+ models produce text alongside tool calls)
response_content = getattr(final_response.choices[0].message, 'content', None)
@@ -1841,12 +1835,10 @@ class OpenAIClient:
if not final_response:
return None
# Check for tool calls
if not final_response.choices or final_response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
tool_calls = getattr(final_response.choices[0].message, 'tool_calls', None)
# Emit llm_content for intermediate narrative display
# (gpt-4.1+ models produce text alongside tool calls)
response_content = getattr(final_response.choices[0].message, 'content', None)
@@ -2045,8 +2037,6 @@ class OpenAIClient:
return
# Check for tool calls
if not final_response.choices or final_response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
tool_calls = getattr(final_response.choices[0].message, 'tool_calls', None)
if tool_calls and execute_tool_fn:
@@ -7,7 +7,7 @@ import threading
from typing import Any, Dict, List, Optional, Union, Literal
import logging
from praisonaiagents._logging import get_logger
from datetime import datetime, timezone
from datetime import datetime
# Decomposed memory functionality - imported as mixins for backward compatibility
from .storage import StorageMixin
@@ -692,7 +692,7 @@ class Memory(StorageMixin, SearchMixin, MemoryCoreMixin):
"_id": ident,
"content": text,
"metadata": metadata,
"created_at": datetime.now(timezone.utc),
"created_at": datetime.utcnow(),
"memory_type": "short_term"
}
self.mongo_short_term.insert_one(doc)
@@ -918,7 +918,7 @@ class Memory(StorageMixin, SearchMixin, MemoryCoreMixin):
"_id": ident,
"content": text,
"metadata": metadata,
"created_at": datetime.now(timezone.utc),
"created_at": datetime.utcnow(),
"memory_type": "long_term"
}
@@ -1441,14 +1441,14 @@ class Memory(StorageMixin, SearchMixin, MemoryCoreMixin):
self.mem0_client.add(text, user_id=user_id, metadata=meta)
elif self.use_mongodb and hasattr(self, "mongo_users"):
try:
from datetime import datetime, timezone
from datetime import datetime
ident = str(time.time_ns())
doc = {
"_id": ident,
"user_id": user_id,
"content": text,
"metadata": meta,
"created_at": datetime.now(timezone.utc)
"created_at": datetime.utcnow()
}
self.mongo_users.insert_one(doc)
self._log_verbose(f"Successfully stored user memory for {user_id}")
@@ -6,13 +6,7 @@ No heavy imports - only stdlib.
"""
import hashlib
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
if TYPE_CHECKING:
from ..knowledge.models import SearchResultItem
ResultItem = Union[Dict[str, Any], "SearchResultItem"]
from typing import Any, Dict, List, Optional, Set, Tuple
def _estimate_tokens(text: str) -> int:
@@ -25,39 +19,6 @@ def _estimate_tokens(text: str) -> int:
return len(text) // 4 + 1
def _extract_value(item: ResultItem, key: str, default: Any = None) -> Any:
"""
Extract value from either dict or object (e.g., SearchResultItem).
Args:
item: Dictionary or object with attributes
key: Key/attribute name to extract
default: Default value if key not found
Returns:
Extracted value or default
"""
if isinstance(item, dict):
return item.get(key, default)
else:
return getattr(item, key, default)
def _extract_metadata_value(
item: ResultItem,
metadata: Dict[str, Any],
key: str,
default: Any = "",
) -> Any:
"""
Extract metadata value with fallback to top-level item attribute.
"""
value = metadata.get(key)
if value is None:
return _extract_value(item, key, default)
return value
def _chunk_hash(text: str, source: Optional[str] = None) -> str:
"""
Generate stable hash for chunk deduplication.
@@ -74,38 +35,31 @@ def _chunk_hash(text: str, source: Optional[str] = None) -> str:
def deduplicate_chunks(
results: List[ResultItem],
results: List[Dict[str, Any]],
similarity_threshold: float = 0.9,
) -> List[ResultItem]:
) -> List[Dict[str, Any]]:
"""
Deduplicate chunks by content hash.
Handles both dict format and SearchResultItem objects.
Args:
results: List of search results (dicts or SearchResultItem objects)
results: List of search results with 'text' or 'memory' key
similarity_threshold: Not used currently (hash-based dedup)
Returns:
Deduplicated list of results
"""
seen_hashes: Set[str] = set()
unique_results: List[ResultItem] = []
unique_results: List[Dict[str, Any]] = []
for result in results:
# Skip None results
if result is None:
continue
# Handle different result formats (dict or SearchResultItem)
text = _extract_value(result, "text") or _extract_value(result, "memory", "")
# Handle different result formats
text = result.get("text") or result.get("memory", "")
# CRITICAL: Handle metadata=None from mem0 - ensure always dict
metadata = _extract_value(result, "metadata") or {}
if not isinstance(metadata, dict):
metadata = {}
source = _extract_metadata_value(result, metadata, "source", "")
metadata = result.get("metadata") or {}
source = metadata.get("source", "")
chunk_id = _chunk_hash(text, source)
@@ -152,19 +106,17 @@ def truncate_context(
def build_context(
results: List[ResultItem],
results: List[Dict[str, Any]],
max_tokens: int = 4000,
deduplicate: bool = True,
separator: str = "\n\n---\n\n",
include_source: bool = True,
) -> Tuple[str, List[ResultItem]]:
) -> Tuple[str, List[Dict[str, Any]]]:
"""
Build context string from retrieval results.
Handles both dict format and SearchResultItem objects.
Args:
results: List of search results (dicts or SearchResultItem objects)
results: List of search results
max_tokens: Maximum tokens for context
deduplicate: Whether to deduplicate chunks
separator: Separator between chunks
@@ -181,7 +133,7 @@ def build_context(
results = deduplicate_chunks(results)
context_parts: List[str] = []
used_results: List[ResultItem] = []
used_results: List[Dict[str, Any]] = []
current_tokens = 0
separator_tokens = _estimate_tokens(separator)
@@ -189,21 +141,17 @@ def build_context(
# Skip None results
if result is None:
continue
# Handle different result formats (dict or SearchResultItem)
text = _extract_value(result, "text") or _extract_value(result, "memory", "")
# Handle different result formats
text = result.get("text") or result.get("memory", "")
if not text:
continue
# Build chunk text with optional source
if include_source:
# CRITICAL: Handle metadata=None from mem0 - ensure always dict
metadata = _extract_value(result, "metadata") or {}
if not isinstance(metadata, dict):
metadata = {}
source = _extract_metadata_value(result, metadata, "source", "")
filename = _extract_metadata_value(result, metadata, "filename", "")
metadata = result.get("metadata") or {}
source = metadata.get("source", "")
filename = metadata.get("filename", "")
source_label = filename or source or f"Source {i + 1}"
chunk_text = f"[{source_label}]\n{text}"
else:
@@ -246,14 +194,11 @@ class DefaultContextBuilder:
def build(
self,
results: List[ResultItem],
results: List[Dict[str, Any]],
max_tokens: int = 4000,
deduplicate: bool = True,
) -> str:
"""Build context string from results.
Handles both dict format and SearchResultItem objects.
"""
"""Build context string from results."""
context, _ = build_context(
results=results,
max_tokens=max_tokens,
@@ -15,8 +15,6 @@ from .protocols import (
ResourceLimits,
)
from .config import SandboxConfig, SecurityPolicy
from .manager import SandboxManager
from .security import SecurityWarning, check_code_safety, format_warnings, get_security_summary
__all__ = [
"SandboxProtocol",
@@ -24,10 +22,5 @@ __all__ = [
"SandboxStatus",
"ResourceLimits",
"SandboxConfig",
"SecurityPolicy",
"SandboxManager",
"SecurityWarning",
"check_code_safety",
"format_warnings",
"get_security_summary",
"SecurityPolicy",
]
@@ -1,302 +0,0 @@
"""
SandboxManager for PraisonAI Agents.
Factory and context manager for sandbox backends.
Core SDK component that routes to appropriate sandbox implementations
in the praisonai wrapper package.
"""
from __future__ import annotations
import logging
from typing import Dict, Any, Optional, TYPE_CHECKING
from .config import SandboxConfig
from .protocols import SandboxProtocol, SandboxResult
if TYPE_CHECKING:
from typing import AsyncContextManager
logger = logging.getLogger(__name__)
class SandboxManager:
"""Factory and context manager for all sandbox backends.
Routes to appropriate sandbox implementations based on configuration.
Lightweight manager in core SDK - heavy implementations in wrapper.
Example:
from praisonaiagents.sandbox import SandboxManager, SandboxConfig
# Simple usage
config = SandboxConfig.docker("python:3.11-slim")
manager = SandboxManager(config)
result = await manager.run_code("print('Hello, World!')")
# Context manager usage
async with SandboxManager(config) as sandbox:
result = await sandbox.execute("print('Hello, World!')")
"""
def __init__(self, config: Optional[SandboxConfig] = None):
"""Initialize the sandbox manager.
Args:
config: Sandbox configuration. Defaults to subprocess sandbox.
"""
self.config = config or SandboxConfig.subprocess()
self._sandbox: Optional[SandboxProtocol] = None
async def __aenter__(self) -> SandboxProtocol:
"""Async context manager entry."""
self._sandbox = await self._create_sandbox()
return self._sandbox
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit with cleanup."""
if self._sandbox:
try:
await self._sandbox.stop()
await self._sandbox.cleanup()
except Exception as e:
logger.warning(f"Error during sandbox cleanup: {e}")
finally:
self._sandbox = None
async def run_code(
self,
code: str,
language: str = "python",
**kwargs
) -> SandboxResult:
"""Convenience method: create sandbox, run code, cleanup.
Args:
code: Code to execute
language: Programming language
**kwargs: Additional arguments passed to execute()
Returns:
SandboxResult with execution details
"""
async with self as sandbox:
return await sandbox.execute(code, language=language, **kwargs)
async def _create_sandbox(self) -> SandboxProtocol:
"""Create appropriate sandbox backend based on config.
Routes to implementations in praisonai wrapper package.
Returns:
Configured sandbox instance
Raises:
ValueError: For unknown sandbox types
ImportError: If required dependencies not available
"""
sandbox_type = self.config.sandbox_type.lower()
if sandbox_type == "docker":
return await self._create_docker_sandbox()
elif sandbox_type in ("subprocess", "local"):
return await self._create_subprocess_sandbox()
elif sandbox_type == "e2b":
return await self._create_e2b_sandbox()
elif sandbox_type == "sandlock":
return await self._create_sandlock_sandbox()
elif sandbox_type == "ssh":
return await self._create_ssh_sandbox()
elif sandbox_type == "modal":
return await self._create_modal_sandbox()
elif sandbox_type == "daytona":
return await self._create_daytona_sandbox()
else:
raise ValueError(
f"Unknown sandbox type: {sandbox_type!r}. "
f"Supported: 'docker', 'subprocess', 'e2b', 'sandlock', 'ssh', 'modal', 'daytona'"
)
async def _create_docker_sandbox(self) -> SandboxProtocol:
"""Create Docker sandbox."""
try:
# Lazy import from wrapper package
from praisonai.sandbox import DockerSandbox
except ImportError as e:
raise ImportError(
"Docker sandbox not available. Install with: "
"pip install praisonaiagents[sandbox-docker]"
) from e
sandbox = DockerSandbox(
image=self.config.image,
config=self.config,
)
if not sandbox.is_available:
raise RuntimeError(
"Docker is not available. Please install Docker and ensure it's running."
)
await sandbox.start()
return sandbox
async def _create_subprocess_sandbox(self) -> SandboxProtocol:
"""Create subprocess sandbox."""
try:
from praisonai.sandbox import SubprocessSandbox
except ImportError as e:
raise ImportError(
"Subprocess sandbox not available. This should not happen - "
"please check your installation."
) from e
sandbox = SubprocessSandbox(config=self.config)
await sandbox.start()
return sandbox
async def _create_e2b_sandbox(self) -> SandboxProtocol:
"""Create E2B sandbox."""
try:
from praisonai.sandbox import E2BSandbox
except ImportError as e:
raise ImportError(
"E2B sandbox not available. Install with: "
"pip install praisonaiagents[sandbox] e2b-code-interpreter"
) from e
sandbox = E2BSandbox(config=self.config)
if not sandbox.is_available:
raise RuntimeError(
"E2B is not available. Please set E2B_API_KEY environment variable."
)
await sandbox.start()
return sandbox
async def _create_sandlock_sandbox(self) -> SandboxProtocol:
"""Create Sandlock sandbox."""
try:
from praisonai.sandbox import SandlockSandbox
except ImportError as e:
raise ImportError(
"Sandlock sandbox not available. Install sandlock."
) from e
sandbox = SandlockSandbox(config=self.config)
await sandbox.start()
return sandbox
async def _create_ssh_sandbox(self) -> SandboxProtocol:
"""Create SSH sandbox."""
try:
from praisonai.sandbox import SSHSandbox
except ImportError as e:
raise ImportError(
"SSH sandbox not available. Install with: "
"pip install paramiko"
) from e
sandbox = SSHSandbox(config=self.config)
await sandbox.start()
return sandbox
async def _create_modal_sandbox(self) -> SandboxProtocol:
"""Create Modal sandbox."""
try:
from praisonai.sandbox import ModalSandbox
except ImportError as e:
raise ImportError(
"Modal sandbox not available. Install with: "
"pip install modal"
) from e
sandbox = ModalSandbox(config=self.config)
await sandbox.start()
return sandbox
async def _create_daytona_sandbox(self) -> SandboxProtocol:
"""Create Daytona sandbox."""
try:
from praisonai.sandbox import DaytonaSandbox
except ImportError as e:
raise ImportError(
"Daytona sandbox not available. Install daytona client."
) from e
sandbox = DaytonaSandbox(config=self.config)
await sandbox.start()
return sandbox
def get_available_types(self) -> Dict[str, Dict[str, Any]]:
"""Get available sandbox types and their status.
Returns:
Dictionary mapping sandbox types to availability info
"""
types = {}
# Check Docker
try:
from praisonai.sandbox import DockerSandbox
docker = DockerSandbox()
types["docker"] = {
"available": docker.is_available,
"description": "Isolated Docker containers",
"requires": ["docker"],
}
except ImportError:
types["docker"] = {
"available": False,
"description": "Isolated Docker containers",
"requires": ["docker", "praisonaiagents[sandbox-docker]"],
}
# Check subprocess (always available)
types["subprocess"] = {
"available": True,
"description": "Local subprocess (limited isolation)",
"requires": [],
}
# Check E2B
try:
from praisonai.sandbox import E2BSandbox
e2b = E2BSandbox()
types["e2b"] = {
"available": e2b.is_available,
"description": "E2B cloud sandboxes",
"requires": ["e2b-code-interpreter", "E2B_API_KEY"],
}
except ImportError:
types["e2b"] = {
"available": False,
"description": "E2B cloud sandboxes",
"requires": ["e2b-code-interpreter", "E2B_API_KEY"],
}
# Add other types with try/except blocks
for sandbox_type, module_name, description, requirements in [
("sandlock", "SandlockSandbox", "OS-native sandboxing", ["sandlock"]),
("ssh", "SSHSandbox", "Remote SSH execution", ["paramiko"]),
("modal", "ModalSandbox", "Modal cloud compute", ["modal"]),
("daytona", "DaytonaSandbox", "Daytona workspaces", ["daytona"]),
]:
try:
from praisonai import sandbox
sandbox_class = getattr(sandbox, module_name)
# Check if class exists without instantiating (some need required args)
types[sandbox_type] = {
"available": False, # Can't easily check without required args
"description": description,
"requires": requirements,
}
except (ImportError, AttributeError):
types[sandbox_type] = {
"available": False,
"description": description,
"requires": requirements,
}
return types
@@ -1,345 +0,0 @@
"""
Security pre-checks for sandbox code execution.
Provides pre-execution analysis of code for potentially dangerous patterns.
Note: These are warnings only - the sandbox provides the real isolation.
"""
import re
import ast
import logging
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class SecurityWarning:
"""A security warning from code analysis."""
pattern: str
message: str
severity: str # "low", "medium", "high", "critical"
line_number: Optional[int] = None
context: Optional[str] = None
# Dangerous patterns to detect in code
DANGEROUS_PATTERNS = [
# System/OS access
(r"import\s+os\s*;?\s*os\.system", "Direct system command execution via os.system", "high"),
(r"subprocess\.call.*shell\s*=\s*True", "Shell injection risk with subprocess", "high"),
(r"subprocess\.run.*shell\s*=\s*True", "Shell injection risk with subprocess", "high"),
(r"subprocess\.Popen.*shell\s*=\s*True", "Shell injection risk with subprocess", "high"),
(r"__import__\(['\"]os['\"]\)", "Dynamic import of os module", "medium"),
(r"eval\s*\(", "Dynamic code execution with eval()", "critical"),
(r"exec\s*\(", "Dynamic code execution with exec()", "critical"),
(r"compile\s*\(.*exec", "Code compilation for execution", "high"),
# File system access
(r"open\s*\(.*['\"]\/etc", "Access to system configuration files", "medium"),
(r"open\s*\(.*['\"]\/root", "Access to root directory", "high"),
(r"open\s*\(.*['\"]\/home", "Access to user directories", "medium"),
(r"shutil\.rmtree", "Recursive directory deletion", "high"),
(r"os\.remove\s*\(", "File deletion", "medium"),
(r"os\.unlink\s*\(", "File deletion", "medium"),
(r"pathlib\.Path.*unlink", "File deletion via pathlib", "medium"),
# Network access
(r"socket\.socket", "Direct socket creation", "medium"),
(r"urllib\.request", "HTTP requests via urllib", "low"),
(r"requests\.", "HTTP requests via requests library", "low"),
(r"http\.client", "HTTP client usage", "low"),
# Process manipulation
(r"os\.fork\s*\(", "Process forking", "high"),
(r"multiprocessing\.", "Multiprocessing usage", "medium"),
(r"threading\.Thread", "Thread creation", "low"),
# Potentially dangerous builtins
(r"getattr\s*\(.*__", "Dynamic attribute access to dunder methods", "medium"),
(r"setattr\s*\(.*__", "Dynamic attribute setting of dunder methods", "medium"),
(r"hasattr\s*\(.*__", "Dynamic attribute checking of dunder methods", "low"),
(r"globals\s*\(\)", "Access to global namespace", "medium"),
(r"locals\s*\(\)", "Access to local namespace", "low"),
(r"vars\s*\(", "Variable inspection", "low"),
(r"dir\s*\(", "Object inspection", "low"),
# Code modification
(r"sys\.modules", "Module manipulation", "high"),
(r"importlib\.", "Dynamic import manipulation", "medium"),
(r"__builtins__", "Access to builtin functions", "high"),
# Memory/performance
(r"while\s+True:", "Infinite loop (potential DoS)", "medium"),
(r"for.*in.*range\s*\(\s*[0-9]{6,}", "Large range iteration", "low"),
(r"list\s*\(\s*range\s*\(\s*[0-9]{6,}", "Large list creation", "low"),
(r"\*\s*[0-9]{6,}", "Large data multiplication", "low"),
]
def check_code_safety(code: str, language: str = "python") -> List[SecurityWarning]:
"""Analyze code for potentially dangerous patterns.
Args:
code: Code to analyze
language: Programming language (currently only 'python' supported)
Returns:
List of security warnings found
Note:
These are warnings only. The sandbox provides real isolation.
This is a best-effort static analysis for user awareness.
"""
warnings = []
if language == "python":
warnings.extend(_check_python_patterns(code))
warnings.extend(_check_python_ast(code))
elif language == "bash":
warnings.extend(_check_bash_patterns(code))
else:
# Fall back to basic pattern matching
warnings.extend(_check_generic_patterns(code))
return warnings
def _check_python_patterns(code: str) -> List[SecurityWarning]:
"""Check Python code using regex patterns."""
warnings = []
lines = code.split('\n')
for i, line in enumerate(lines, 1):
for pattern, message, severity in DANGEROUS_PATTERNS:
if re.search(pattern, line, re.IGNORECASE):
warnings.append(SecurityWarning(
pattern=pattern,
message=message,
severity=severity,
line_number=i,
context=line.strip(),
))
return warnings
def _check_python_ast(code: str) -> List[SecurityWarning]:
"""Check Python code using AST analysis."""
warnings = []
try:
tree = ast.parse(code)
visitor = DangerousASTVisitor()
visitor.visit(tree)
warnings.extend(visitor.warnings)
except SyntaxError:
# Invalid Python syntax - let the sandbox handle it
pass
except Exception as e:
logger.debug(f"AST analysis failed: {e}")
return warnings
def _check_bash_patterns(code: str) -> List[SecurityWarning]:
"""Check bash/shell code for dangerous patterns."""
warnings = []
lines = code.split('\n')
bash_patterns = [
(r"rm\s+-rf\s+\/", "Recursive deletion of root directory", "critical"),
(r"rm\s+-rf\s+\*", "Recursive deletion with wildcard", "high"),
(r">\s*\/dev\/sd[a-z]", "Direct disk device access", "critical"),
(r"dd\s+if=.*of=", "Low-level disk operations", "high"),
(r"mkfs\.", "Filesystem creation", "critical"),
(r"fdisk", "Disk partitioning", "critical"),
(r"shutdown", "System shutdown", "high"),
(r"reboot", "System reboot", "high"),
(r"halt", "System halt", "high"),
(r"init\s+[0-6]", "System runlevel change", "high"),
(r"chmod\s+777", "Overly permissive file permissions", "medium"),
(r"curl.*\|\s*bash", "Piped execution from network", "high"),
(r"wget.*\|\s*bash", "Piped execution from network", "high"),
(r"\$\(.*\)", "Command substitution", "medium"),
(r"`.*`", "Command substitution (backticks)", "medium"),
]
for i, line in enumerate(lines, 1):
for pattern, message, severity in bash_patterns:
if re.search(pattern, line, re.IGNORECASE):
warnings.append(SecurityWarning(
pattern=pattern,
message=message,
severity=severity,
line_number=i,
context=line.strip(),
))
return warnings
def _check_generic_patterns(code: str) -> List[SecurityWarning]:
"""Check code using generic patterns."""
warnings = []
lines = code.split('\n')
generic_patterns = [
(r"eval\s*\(", "Dynamic code execution", "critical"),
(r"exec\s*\(", "Dynamic code execution", "critical"),
(r"system\s*\(", "System command execution", "high"),
(r"shell\s*\(", "Shell command execution", "high"),
]
for i, line in enumerate(lines, 1):
for pattern, message, severity in generic_patterns:
if re.search(pattern, line, re.IGNORECASE):
warnings.append(SecurityWarning(
pattern=pattern,
message=message,
severity=severity,
line_number=i,
context=line.strip(),
))
return warnings
class DangerousASTVisitor(ast.NodeVisitor):
"""AST visitor to detect dangerous Python constructs."""
def __init__(self):
self.warnings = []
def visit_Call(self, node):
"""Visit function calls."""
# Check for dangerous function calls
if isinstance(node.func, ast.Name):
func_name = node.func.id
if func_name in ("eval", "exec", "compile"):
self.warnings.append(SecurityWarning(
pattern=f"{func_name}()",
message=f"Dynamic code execution with {func_name}()",
severity="critical",
line_number=node.lineno,
))
# Check for attribute calls like os.system
elif isinstance(node.func, ast.Attribute):
if isinstance(node.func.value, ast.Name):
module = node.func.value.id
func = node.func.attr
if module == "os" and func == "system":
self.warnings.append(SecurityWarning(
pattern="os.system()",
message="Direct system command execution",
severity="high",
line_number=node.lineno,
))
elif module == "subprocess" and func in ("call", "run", "Popen"):
# Check if shell=True is used
for keyword in node.keywords:
if keyword.arg == "shell" and isinstance(keyword.value, ast.Constant):
if keyword.value.value is True:
self.warnings.append(SecurityWarning(
pattern=f"subprocess.{func}(..., shell=True)",
message="Shell injection risk with subprocess",
severity="high",
line_number=node.lineno,
))
self.generic_visit(node)
def visit_Import(self, node):
"""Visit import statements."""
for alias in node.names:
if alias.name in ("os", "subprocess", "sys"):
self.warnings.append(SecurityWarning(
pattern=f"import {alias.name}",
message=f"Import of potentially dangerous module: {alias.name}",
severity="low",
line_number=node.lineno,
))
self.generic_visit(node)
def visit_ImportFrom(self, node):
"""Visit from ... import statements."""
if node.module in ("os", "subprocess", "sys"):
for alias in node.names:
if alias.name in ("system", "call", "run", "Popen", "modules"):
self.warnings.append(SecurityWarning(
pattern=f"from {node.module} import {alias.name}",
message=f"Import of dangerous function: {alias.name}",
severity="medium",
line_number=node.lineno,
))
self.generic_visit(node)
def format_warnings(warnings: List[SecurityWarning]) -> str:
"""Format security warnings for display.
Args:
warnings: List of security warnings
Returns:
Formatted string of warnings
"""
if not warnings:
return "No security issues detected."
# Group by severity
by_severity = {"critical": [], "high": [], "medium": [], "low": []}
for warning in warnings:
by_severity[warning.severity].append(warning)
lines = []
lines.append(f"Security analysis found {len(warnings)} potential issue(s):")
lines.append("")
for severity in ["critical", "high", "medium", "low"]:
if by_severity[severity]:
lines.append(f"{severity.upper()} RISK:")
for warning in by_severity[severity]:
line_info = f" (line {warning.line_number})" if warning.line_number else ""
context_info = f"\n Context: {warning.context}" if warning.context else ""
lines.append(f" - {warning.message}{line_info}{context_info}")
lines.append("")
lines.append("Note: These are warnings only. The sandbox provides real isolation.")
return "\n".join(lines)
def get_security_summary(warnings: List[SecurityWarning]) -> Dict[str, Any]:
"""Get security analysis summary.
Args:
warnings: List of security warnings
Returns:
Dictionary with summary statistics
"""
by_severity = {"critical": 0, "high": 0, "medium": 0, "low": 0}
for warning in warnings:
by_severity[warning.severity] += 1
max_severity = "low"
if by_severity["critical"] > 0:
max_severity = "critical"
elif by_severity["high"] > 0:
max_severity = "high"
elif by_severity["medium"] > 0:
max_severity = "medium"
return {
"total_warnings": len(warnings),
"by_severity": by_severity,
"max_severity": max_severity,
"is_safe": len(warnings) == 0,
"has_critical": by_severity["critical"] > 0,
}
@@ -1,29 +0,0 @@
# Scheduler — runner vs loop
## ScheduleRunner (stateless)
`ScheduleRunner` is **stateless**. Call `get_due_jobs()` whenever you want to check for jobs that should fire. It does not run background threads.
Use when:
- CLI one-shot tick (`praisonai schedule tick`)
- Host integration that polls on its own timer
- Tests
## ScheduleLoop (optional daemon)
`ScheduleLoop` is an **optional background daemon**. You must call `.start()` explicitly; it polls the runner and invokes `on_trigger` callbacks.
Use when:
- Long-running host process (Pattern B/C via `setup_bridges()`)
- Standalone scheduler service
```python
from praisonaiagents.scheduler import FileScheduleStore, ScheduleRunner, ScheduleLoop
store = FileScheduleStore()
runner = ScheduleRunner(store)
loop = ScheduleLoop(runner)
loop.start() # optional daemon
```
The wrapper host calls `ensure_schedule_runner()` which starts `ScheduleLoop` when the scheduler optional dependency is available.
@@ -32,22 +32,7 @@ When you provide a `session_id` to an Agent:
### Default Storage Location
Sessions are stored in: `~/.praisonai/sessions/{session_id}.json`
### Session metadata contract (Issue #48)
List endpoints and UI dashboards may expose these optional root-level fields:
| Field | Type | Description |
|-------|------|-------------|
| `model` | string | LLM model used in the session |
| `total_tokens` | int | Cumulative input+output tokens |
| `cost` | float | Estimated USD cost |
| `agent_id` | string | Gateway or registry agent id |
| `source` | string | Origin: `chat`, `gateway`, `cli`, `api` |
| `agent_name` | string | Human-readable agent name |
Fields are persisted in session JSON `metadata` and mirrored at root on save. Populated automatically after each assistant turn when using the default session store.
Sessions are stored in: `~/.praison/sessions/{session_id}.json`
### Session File Format
@@ -60,13 +45,7 @@ Fields are persisted in session JSON `metadata` and mirrored at root on save. Po
],
"created_at": "2026-01-02T04:00:00+00:00",
"updated_at": "2026-01-02T04:01:00+00:00",
"agent_name": "Assistant",
"agent_id": "agent-abc",
"source": "chat",
"model": "gpt-4o-mini",
"total_tokens": 128,
"cost": 0.0004,
"metadata": {}
"agent_name": "Assistant"
}
```
@@ -144,35 +144,6 @@ class HierarchicalSessionStore(DefaultSessionStore):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._extended_cache: Dict[str, ExtendedSessionData] = {}
def _load_session_from_disk(self, session_id: str, filepath: str) -> ExtendedSessionData:
"""Load extended session JSON from disk (caller must hold FileLock)."""
if os.path.exists(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return ExtendedSessionData.from_dict(data)
except (json.JSONDecodeError, IOError):
pass
return ExtendedSessionData(session_id=session_id)
def _modify_session_locked(
self,
session_id: str,
mutator,
*,
error_label: str = "modify session",
) -> bool:
"""Locked read-modify-write preserving extended session fields."""
result = super()._modify_session_locked(
session_id, mutator, error_label=error_label
)
if result:
with self._lock:
cached = self._cache.get(session_id)
if isinstance(cached, ExtendedSessionData):
self._extended_cache[session_id] = cached
return result
def add_message(
self,
@@ -186,22 +157,25 @@ class HierarchicalSessionStore(DefaultSessionStore):
Overrides parent to preserve extended session data.
"""
from .store import SessionMessage
# Load extended session (force reload to get latest)
session = self._load_extended_session(session_id, force_reload=True)
message = SessionMessage(
role=role,
content=content,
timestamp=time.time(),
metadata=metadata or {},
)
def _apply(session: SessionData) -> None:
session.messages.append(message)
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages :]
return self._modify_session_locked(
session_id, _apply, error_label="add message to session"
)
session.messages.append(message)
# Trim messages if over limit
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages:]
return self._save_extended_session(session)
def _load_extended_session(self, session_id: str, force_reload: bool = False) -> ExtendedSessionData:
"""Load extended session from disk."""
@@ -129,16 +129,3 @@ class SessionStoreProtocol(Protocol):
True if the session exists.
"""
...
@runtime_checkable
class CheckpointQueryProtocol(Protocol):
"""Read path for session checkpoints / rollback snapshots."""
def list_checkpoints(self, session_id: str) -> List[Dict[str, Any]]:
"""Return checkpoint metadata for a session."""
...
def get_checkpoint(self, session_id: str, checkpoint_id: str) -> Optional[Dict[str, Any]]:
"""Return a single checkpoint payload."""
...
@@ -22,7 +22,7 @@ else:
_HAS_FCNTL = False
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Dict, List, Optional
from ..paths import get_sessions_dir
@@ -81,7 +81,7 @@ class SessionData:
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
data = {
return {
"session_id": self.session_id,
"messages": [m.to_dict() for m in self.messages],
"created_at": self.created_at,
@@ -92,10 +92,6 @@ class SessionData:
"gateway_session_id": self.gateway_session_id,
"agent_id": self.agent_id,
}
for key in ("model", "llm", "total_tokens", "token_count", "cost", "source"):
if key in self.metadata:
data[key] = self.metadata[key]
return data
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SessionData":
@@ -141,8 +137,7 @@ class FileLock:
self._lock_path = filepath + ".lock"
def __enter__(self):
if not self.acquire():
raise IOError(f"Failed to acquire file lock for {self.filepath} after {self.timeout}s")
self.acquire()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
@@ -291,72 +286,6 @@ class DefaultSessionStore:
self._cache[session_id] = session
return session
def _load_session_from_disk(self, session_id: str, filepath: str) -> SessionData:
"""Load session JSON from disk (caller must hold FileLock)."""
if os.path.exists(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return SessionData.from_dict(data)
except (json.JSONDecodeError, IOError):
pass
return SessionData(session_id=session_id)
def _read_session_fresh(self, session_id: str) -> SessionData:
"""Reload session from disk and refresh the in-process cache."""
filepath = self._get_session_path(session_id)
with FileLock(filepath, self.lock_timeout):
session = self._load_session_from_disk(session_id, filepath)
with self._lock:
self._cache[session_id] = session
return session
def _modify_session_locked(
self,
session_id: str,
mutator: Callable[[SessionData], None],
*,
error_label: str = "modify session",
) -> bool:
"""Apply mutator after reloading from disk under FileLock."""
filepath = self._get_session_path(session_id)
with FileLock(filepath, self.lock_timeout):
session = self._load_session_from_disk(session_id, filepath)
mutator(session)
session.updated_at = datetime.now(timezone.utc).isoformat()
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages:]
try:
dir_path = os.path.dirname(filepath) or "."
os.makedirs(dir_path, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=dir_path,
delete=False,
suffix=".tmp",
) as f:
json.dump(session.to_dict(), f, indent=2, ensure_ascii=False)
temp_path = f.name
os.replace(temp_path, filepath)
with self._lock:
self._cache[session_id] = session
return True
except (IOError, OSError) as e:
logger.error(f"Failed to {error_label} {session_id}: {e}")
try:
if "temp_path" in locals():
os.remove(temp_path)
except (IOError, OSError):
pass
return False
def _save_session(self, session: SessionData) -> bool:
"""Save session to disk with atomic write."""
@@ -506,13 +435,13 @@ class DefaultSessionStore:
Returns:
List of {"role": "user/assistant", "content": "..."} dicts.
"""
session = self._read_session_fresh(session_id)
session = self._load_session(session_id)
limit = max_messages or self.max_messages
return session.get_chat_history(limit)
def get_session(self, session_id: str) -> SessionData:
"""Get full session data."""
return self._read_session_fresh(session_id)
return self._load_session(session_id)
def set_agent_info(
self,
@@ -521,42 +450,27 @@ class DefaultSessionStore:
user_id: Optional[str] = None,
) -> bool:
"""Set agent info for a session."""
def _apply(session: SessionData) -> None:
session = self._load_session(session_id)
with self._lock:
if agent_name:
session.agent_name = agent_name
if user_id:
session.user_id = user_id
return self._modify_session_locked(
session_id, _apply, error_label="set agent info for session"
)
self._cache[session_id] = session
return self._save_session(session)
def clear_session(self, session_id: str) -> bool:
"""Clear all messages from a session."""
return self._modify_session_locked(
session_id,
lambda session: session.messages.clear(),
error_label="clear session",
)
def update_session_metadata(self, session_id: str, **fields: Any) -> bool:
"""Merge run stats / metadata fields into a persisted session."""
if not fields:
return True
def _apply(session: SessionData) -> None:
for key, value in fields.items():
if value is None:
continue
session.metadata[key] = value
if key in ("agent_id", "agent_name", "user_id"):
setattr(session, key, value)
return self._modify_session_locked(
session_id, _apply, error_label="update session metadata"
)
session = self._load_session(session_id)
with self._lock:
session.messages.clear()
self._cache[session_id] = session
return self._save_session(session)
def delete_session(self, session_id: str) -> bool:
"""Delete a session completely."""
filepath = self._get_session_path(session_id)
@@ -586,16 +500,10 @@ class DefaultSessionStore:
data = json.load(f)
sessions.append({
"session_id": data.get("session_id", filename[:-5]),
"id": data.get("session_id", filename[:-5]),
"agent_name": data.get("agent_name"),
"agent_id": data.get("agent_id") or (data.get("metadata") or {}).get("agent_id"),
"source": data.get("source") or (data.get("metadata") or {}).get("source"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"message_count": len(data.get("messages", [])),
"model": data.get("model") or data.get("llm") or (data.get("metadata") or {}).get("model"),
"total_tokens": data.get("total_tokens") or data.get("token_count") or (data.get("metadata") or {}).get("total_tokens"),
"cost": data.get("cost") or (data.get("metadata") or {}).get("cost"),
})
except (json.JSONDecodeError, IOError):
continue
@@ -655,7 +563,7 @@ class DefaultSessionStore:
List of SessionData objects for the specified agent
"""
session_ids = self.list_sessions_by_agent(agent_name, limit)
return [self._read_session_fresh(sid) for sid in session_ids]
return [self._load_session(sid) for sid in session_ids]
def get_agent_chat_history(
self,
@@ -709,15 +617,16 @@ class DefaultSessionStore:
Returns:
True if saved successfully
"""
def _apply(session: SessionData) -> None:
session = self._load_session(session_id)
with self._lock:
if gateway_session_id:
session.gateway_session_id = gateway_session_id
if agent_id:
session.agent_id = agent_id
return self._modify_session_locked(
session_id, _apply, error_label="set gateway info for session"
)
self._cache[session_id] = session
return self._save_session(session)
def get_by_gateway_session(self, gateway_session_id: str) -> Optional[SessionData]:
"""Get session data linked to a gateway session.
@@ -19,9 +19,7 @@ Usage:
__all__ = [
# Models
"SkillProperties",
"SkillMetadata",
"SkillRequirements",
"SkillState",
"SkillMetadata",
"ParseError",
"ValidationError",
# Parser
@@ -56,17 +54,13 @@ __all__ = [
"SkillActivatedEvent",
# Budget
"SkillPromptBudget",
# Capability validation
"CapabilityValidator",
"EnforcementLevel",
"ValidationResult",
]
def __getattr__(name: str):
"""Lazy load module components to avoid import overhead."""
if name in ("SkillProperties", "SkillMetadata", "SkillRequirements", "SkillState", "ParseError", "ValidationError"):
from .models import SkillProperties, SkillMetadata, SkillRequirements, SkillState, ParseError, ValidationError
if name in ("SkillProperties", "SkillMetadata", "ParseError", "ValidationError"):
from .models import SkillProperties, SkillMetadata, ParseError, ValidationError
return locals()[name]
if name in ("parse_frontmatter", "find_skill_md", "read_properties"):
@@ -136,9 +130,5 @@ def __getattr__(name: str):
return None
return load_skill
if name in ("CapabilityValidator", "EnforcementLevel", "ValidationResult"):
from .capability_validator import CapabilityValidator, EnforcementLevel, ValidationResult
return locals()[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -1,236 +0,0 @@
"""Capability validation for Agent Skills.
This module implements the capability gate enforcement system for skills,
checking that required tools, servers, and environment variables are available
before allowing skill activation.
"""
import logging
import os
from dataclasses import dataclass
from typing import Dict, List, Optional, Set, Tuple
from enum import Enum
from .models import SkillProperties, SkillRequirements, SkillState
logger = logging.getLogger(__name__)
class EnforcementLevel(Enum):
"""Enforcement level for capability gates."""
DISABLED = "disabled" # No enforcement (existing behavior)
TELEMETRY = "telemetry" # Log only, no blocking
WARN = "warn" # Warning but allow activation
STRICT = "strict" # Hard failure, block activation
@dataclass
class ValidationResult:
"""Result of capability validation for a skill."""
skill_name: str
state: SkillState
satisfied_tools: List[str]
missing_tools: List[str]
satisfied_servers: List[str]
missing_servers: List[str]
satisfied_env_vars: List[str]
missing_env_vars: List[str]
warnings: List[str]
errors: List[str]
@property
def is_fully_satisfied(self) -> bool:
"""Check if all requirements are satisfied."""
return not any([
self.missing_tools,
self.missing_servers,
self.missing_env_vars,
self.errors
])
@property
def has_critical_missing(self) -> bool:
"""Check if critical requirements are missing (tools, servers)."""
return bool(self.missing_tools or self.missing_servers)
def to_dict(self) -> Dict:
"""Convert to dictionary for serialization."""
return {
"skill_name": self.skill_name,
"state": self.state.value,
"satisfied_tools": self.satisfied_tools,
"missing_tools": self.missing_tools,
"satisfied_servers": self.satisfied_servers,
"missing_servers": self.missing_servers,
"satisfied_env_vars": self.satisfied_env_vars,
"missing_env_vars": self.missing_env_vars,
"warnings": self.warnings,
"errors": self.errors,
"is_fully_satisfied": self.is_fully_satisfied,
"has_critical_missing": self.has_critical_missing
}
class CapabilityValidator:
"""Validates skill requirements against available capabilities.
This is the core capability gate enforcement engine. It checks skill
requirements against the runtime context (available tools, servers, etc.)
and determines whether a skill should be activated based on policy.
"""
def __init__(self, enforcement_level: EnforcementLevel = EnforcementLevel.WARN):
"""Initialize the validator.
Args:
enforcement_level: How strictly to enforce requirements
"""
self.enforcement_level = enforcement_level
self._tool_cache: Optional[Set[str]] = None
self._server_cache: Optional[Set[str]] = None
def validate_skill(
self,
skill: SkillProperties,
available_tools: Optional[Set[str]] = None,
available_servers: Optional[Set[str]] = None
) -> ValidationResult:
"""Validate a skill's requirements against available capabilities.
Args:
skill: Skill to validate
available_tools: Set of available tool names (auto-detected if None)
available_servers: Set of available server names (auto-detected if None)
Returns:
ValidationResult with validation status and details
"""
if not skill.requirements or skill.requirements.is_empty():
# No requirements specified - skill is always available
return ValidationResult(
skill_name=skill.name,
state=SkillState.ACTIVE,
satisfied_tools=[],
missing_tools=[],
satisfied_servers=[],
missing_servers=[],
satisfied_env_vars=[],
missing_env_vars=[],
warnings=[],
errors=[]
)
# Get available capabilities
if available_tools is None:
available_tools = self._get_available_tools()
if available_servers is None:
available_servers = self._get_available_servers()
requirements = skill.requirements
# Validate tools
satisfied_tools = [tool for tool in requirements.tools if tool in available_tools]
missing_tools = [tool for tool in requirements.tools if tool not in available_tools]
# Validate servers
satisfied_servers = [srv for srv in requirements.servers if srv in available_servers]
missing_servers = [srv for srv in requirements.servers if srv not in available_servers]
# Validate environment variables
satisfied_env_vars = [var for var in requirements.env_vars if os.getenv(var)]
missing_env_vars = [var for var in requirements.env_vars if not os.getenv(var)]
# Determine warnings and errors
warnings = []
errors = []
if missing_tools:
msg = f"Missing required tools: {', '.join(missing_tools)}"
if self.enforcement_level in (EnforcementLevel.STRICT,):
errors.append(msg)
else:
warnings.append(msg)
if missing_servers:
msg = f"Missing required servers: {', '.join(missing_servers)}"
if self.enforcement_level in (EnforcementLevel.STRICT,):
errors.append(msg)
else:
warnings.append(msg)
if missing_env_vars:
msg = f"Missing required environment variables: {', '.join(missing_env_vars)}"
if self.enforcement_level == EnforcementLevel.STRICT:
errors.append(msg)
else:
warnings.append(msg)
# Determine skill state
if errors:
state = SkillState.UNAVAILABLE
elif warnings:
state = SkillState.DEGRADED
else:
state = SkillState.ACTIVE
result = ValidationResult(
skill_name=skill.name,
state=state,
satisfied_tools=satisfied_tools,
missing_tools=missing_tools,
satisfied_servers=satisfied_servers,
missing_servers=missing_servers,
satisfied_env_vars=satisfied_env_vars,
missing_env_vars=missing_env_vars,
warnings=warnings,
errors=errors
)
# Log based on enforcement level
self._log_validation_result(result)
return result
def _get_available_tools(self) -> Set[str]:
"""Get set of available tool names from registry."""
if self._tool_cache is None:
try:
from ..tools.registry import get_registry
registry = get_registry()
self._tool_cache = set(registry.list_tools())
except ImportError:
logger.debug("Tool registry not available")
self._tool_cache = set()
return self._tool_cache
def _get_available_servers(self) -> Set[str]:
"""Get set of available MCP server names."""
if self._server_cache is None:
# TODO: Implement MCP server discovery
# For now, return empty set - this can be extended later
self._server_cache = set()
return self._server_cache
def _log_validation_result(self, result: ValidationResult) -> None:
"""Log validation result based on enforcement level."""
if self.enforcement_level == EnforcementLevel.DISABLED:
return
skill_name = result.skill_name
if result.state == SkillState.UNAVAILABLE:
if self.enforcement_level == EnforcementLevel.STRICT:
logger.error(f"Skill '{skill_name}' blocked: {'; '.join(result.errors)}")
else:
logger.warning(f"Skill '{skill_name}' has critical missing dependencies: {'; '.join(result.errors)}")
elif result.state == SkillState.DEGRADED:
logger.warning(f"Skill '{skill_name}' partially available: {'; '.join(result.warnings)}")
elif result.state == SkillState.ACTIVE:
logger.debug(f"Skill '{skill_name}' requirements satisfied")
def clear_cache(self) -> None:
"""Clear cached capability information."""
self._tool_cache = None
self._server_cache = None
@@ -108,7 +108,7 @@ class SkillLoader:
return False
try:
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
_, body = parse_frontmatter(content)
skill.instructions = body
return True
@@ -136,7 +136,7 @@ class SkillLoader:
for script_file in scripts_dir.iterdir():
if script_file.is_file():
try:
scripts[script_file.name] = script_file.read_text(encoding="utf-8")
scripts[script_file.name] = script_file.read_text()
except Exception:
continue
except PermissionError:
@@ -166,7 +166,7 @@ class SkillLoader:
for ref_file in refs_dir.iterdir():
if ref_file.is_file():
try:
refs[ref_file.name] = ref_file.read_text(encoding="utf-8")
refs[ref_file.name] = ref_file.read_text()
except Exception:
continue
except PermissionError:
@@ -5,13 +5,12 @@ from typing import List, Optional, Dict
logger = logging.getLogger(__name__)
from .models import SkillMetadata, SkillState
from .discovery import discover_skills, get_default_skill_dirs
from .models import SkillMetadata
from .discovery import discover_skills
from .loader import SkillLoader, LoadedSkill
from .prompt import generate_skills_xml
from .substitution import render_skill_body
from .shell_render import render_shell_blocks
from .capability_validator import CapabilityValidator, EnforcementLevel, ValidationResult
class SkillManager:
@@ -35,21 +34,11 @@ class SkillManager:
manager.activate(skill)
"""
def __init__(self, enforcement_level: Optional[EnforcementLevel] = None):
"""Initialize the SkillManager.
Args:
enforcement_level: Capability enforcement level (defaults to environment-based)
"""
def __init__(self):
"""Initialize the SkillManager."""
self._skills: Dict[str, LoadedSkill] = {}
self._loader = SkillLoader()
self._discovered = False
self._validation_cache: Dict[str, ValidationResult] = {}
# Initialize capability validator
if enforcement_level is None:
enforcement_level = self._get_default_enforcement_level()
self._validator = CapabilityValidator(enforcement_level)
@property
def skills(self) -> List[LoadedSkill]:
@@ -137,29 +126,15 @@ class SkillManager:
def get_available_skills(self) -> List[SkillMetadata]:
"""Get metadata for all available skills.
This is used for system prompt injection. Skills are filtered by:
1. disable-model-invocation: true (excluded)
2. Capability enforcement (if strict mode, exclude UNAVAILABLE skills)
This is used for system prompt injection. Skills with
``disable-model-invocation: true`` are omitted so the LLM never
sees them and cannot auto-trigger them.
"""
available = []
for skill in self._skills.values():
# Skip if explicitly disabled for model invocation
if getattr(skill.properties, "disable_model_invocation", False):
continue
# Check capability requirements if enforcement is strict
if self._validator.enforcement_level == EnforcementLevel.STRICT:
try:
validation = self.validate_skill_capabilities(skill.properties.name)
if validation.state == SkillState.UNAVAILABLE:
continue # Skip unavailable skills in strict mode
except Exception as e:
logger.warning(f"Skipping skill '{skill.properties.name}' due to validation error: {e}")
continue
available.append(skill.metadata)
return available
return [
skill.metadata
for skill in self._skills.values()
if not getattr(skill.properties, "disable_model_invocation", False)
]
def get_user_invocable_skills(self) -> List[LoadedSkill]:
"""Return skills the *user* may invoke via slash-commands.
@@ -296,7 +271,8 @@ class SkillManager:
return {"success": False, "error": "Skill content exceeds maximum size (100KB)"}
# Create skill directory
skill_dirs = get_default_skill_dirs()
from .discovery import get_default_skill_directories
skill_dirs = get_default_skill_directories()
base_dir = skill_dirs[0] if skill_dirs else "~/.praisonai/skills"
import os
@@ -591,90 +567,9 @@ author: agent
logger.debug("Failed to clean up temp file %s", temp_path, exc_info=True)
raise
def validate_skill_capabilities(self, skill_name: str, force_refresh: bool = False) -> ValidationResult:
"""Validate a skill's capability requirements.
Args:
skill_name: Name of the skill to validate
force_refresh: If True, bypass cache and re-validate
Returns:
ValidationResult with capability status
Raises:
ValueError: If skill not found
"""
skill = self.get_skill(skill_name)
if skill is None:
raise ValueError(f"Skill '{skill_name}' not found")
# Check cache first
if not force_refresh and skill_name in self._validation_cache:
return self._validation_cache[skill_name]
# Perform validation
result = self._validator.validate_skill(skill.properties)
self._validation_cache[skill_name] = result
return result
def get_available_skills_by_state(self, state: SkillState) -> List[LoadedSkill]:
"""Get skills filtered by their capability validation state.
Args:
state: Desired skill state
Returns:
List of skills in the specified state
"""
matching = []
for skill in self._skills.values():
try:
result = self.validate_skill_capabilities(skill.properties.name)
if result.state == state:
matching.append(skill)
except Exception as e:
logger.warning(f"Failed to validate skill '{skill.properties.name}': {e}")
continue
return matching
def get_skills_diagnostics(self) -> Dict[str, ValidationResult]:
"""Get capability diagnostics for all skills.
Returns:
Dict mapping skill names to their validation results
"""
diagnostics = {}
for skill_name in self.skill_names:
try:
diagnostics[skill_name] = self.validate_skill_capabilities(skill_name)
except Exception as e:
logger.error(f"Failed to validate skill '{skill_name}': {e}")
return diagnostics
def _get_default_enforcement_level(self) -> EnforcementLevel:
"""Get default enforcement level from environment."""
import os
level_str = os.getenv('SKILL_CAPABILITY_ENFORCEMENT', 'warn').lower()
level_map = {
'disabled': EnforcementLevel.DISABLED,
'off': EnforcementLevel.DISABLED,
'telemetry': EnforcementLevel.TELEMETRY,
'log': EnforcementLevel.TELEMETRY,
'warn': EnforcementLevel.WARN,
'warning': EnforcementLevel.WARN,
'strict': EnforcementLevel.STRICT,
'hard': EnforcementLevel.STRICT,
'fail': EnforcementLevel.STRICT,
}
return level_map.get(level_str, EnforcementLevel.WARN)
def clear(self) -> None:
"""Clear all loaded skills."""
self._skills.clear()
self._validation_cache.clear()
self._validator.clear_cache()
self._discovered = False
def __len__(self) -> int:
@@ -1,9 +1,8 @@
"""Data models for Agent Skills."""
from dataclasses import dataclass, field
from typing import Optional, List, Union
from typing import Optional
from pathlib import Path
from enum import Enum
class ParseError(Exception):
@@ -16,105 +15,6 @@ class ValidationError(Exception):
pass
class SkillState(Enum):
"""Skill activation state based on requirement validation."""
ACTIVE = "active" # All requirements satisfied
DEGRADED = "degraded" # Some requirements missing (soft warn)
UNAVAILABLE = "unavailable" # Critical requirements missing (hard fail)
UNKNOWN = "unknown" # Requirements not yet validated
@dataclass
class SkillRequirements:
"""Capability requirements for a skill.
This represents the parsed and normalized requirements from skill frontmatter.
Supports both Hermes/OpenClaw conventions and PraisonAI-specific extensions.
Attributes:
servers: Required MCP servers or service endpoints
tools: Required tool names (from tool registry)
env_vars: Required environment variables (use sparingly)
openclaw_hints: OpenClaw-specific metadata (passthrough)
"""
servers: List[str] = field(default_factory=list)
tools: List[str] = field(default_factory=list)
env_vars: List[str] = field(default_factory=list)
openclaw_hints: dict = field(default_factory=dict)
@classmethod
def from_frontmatter(cls, metadata: dict) -> "SkillRequirements":
"""Parse requirements from skill frontmatter.
Normalizes various naming conventions:
- requires_servers, requires_tools (Hermes/OpenClaw style)
- allowed-tools (existing PraisonAI convention)
- openclaw (OpenClaw-specific hints)
Args:
metadata: Raw frontmatter dict
Returns:
SkillRequirements instance
"""
servers = []
tools = []
env_vars = []
openclaw_hints = {}
# Parse servers
if "requires_servers" in metadata:
servers = cls._normalize_list(metadata["requires_servers"])
elif "requires-servers" in metadata:
servers = cls._normalize_list(metadata["requires-servers"])
# Parse tools (multiple naming conventions)
if "requires_tools" in metadata:
tools.extend(cls._normalize_list(metadata["requires_tools"]))
elif "requires-tools" in metadata:
tools.extend(cls._normalize_list(metadata["requires-tools"]))
# Support existing allowed-tools convention for backward compatibility
if "allowed-tools" in metadata:
tools.extend(cls._normalize_list(metadata["allowed-tools"]))
# Parse environment variables
if "requires_env" in metadata:
env_vars = cls._normalize_list(metadata["requires_env"])
elif "requires-env" in metadata:
env_vars = cls._normalize_list(metadata["requires-env"])
# OpenClaw hints (passthrough)
if "openclaw" in metadata:
openclaw_hints = metadata["openclaw"] if isinstance(metadata["openclaw"], dict) else {}
return cls(
servers=servers,
tools=tools,
env_vars=env_vars,
openclaw_hints=openclaw_hints
)
@staticmethod
def _normalize_list(value: Union[str, List[str]]) -> List[str]:
"""Normalize string or list to list of strings."""
if isinstance(value, str):
# Support both comma-separated and space-separated strings
return [item.strip() for item in value.replace(',', ' ').split() if item.strip()]
elif isinstance(value, (list, tuple)):
return [str(item).strip() for item in value if str(item).strip()]
else:
return []
def is_empty(self) -> bool:
"""Check if no requirements are specified."""
return not any([self.servers, self.tools, self.env_vars, self.openclaw_hints])
def __bool__(self) -> bool:
"""Truth value: True if any requirements are specified."""
return not self.is_empty()
@dataclass
class SkillProperties:
"""Properties parsed from a skill's SKILL.md frontmatter.
@@ -149,8 +49,6 @@ class SkillProperties:
hooks: Optional[dict] = None
paths: Optional[object] = None # list[str] glob patterns
shell: Optional[str] = None # "bash" | "powershell"
# Capability requirements (new)
requirements: Optional[SkillRequirements] = None
def to_dict(self) -> dict:
"""Convert to dictionary, excluding None values and empty metadata."""
@@ -3,7 +3,7 @@
from pathlib import Path
from typing import Optional, Tuple
from .models import ParseError, ValidationError, SkillProperties, SkillRequirements
from .models import ParseError, ValidationError, SkillProperties
def find_skill_md(skill_dir: Path) -> Optional[Path]:
@@ -100,7 +100,7 @@ def read_properties(skill_dir: Path) -> SkillProperties:
if skill_md is None:
raise ParseError(f"SKILL.md not found in {skill_dir}")
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
metadata, _ = parse_frontmatter(content)
if "name" not in metadata:
@@ -143,6 +143,4 @@ def read_properties(skill_dir: Path) -> SkillProperties:
hooks=metadata.get("hooks") if isinstance(metadata.get("hooks"), dict) else None,
paths=metadata.get("paths"),
shell=metadata.get("shell"),
# Parse capability requirements
requirements=SkillRequirements.from_frontmatter(metadata),
)
@@ -7,7 +7,7 @@ plugin bundle, enterprise policy) without touching the core.
from __future__ import annotations
from typing import Protocol, Iterable, Optional, runtime_checkable, List, Dict, Any
from typing import Protocol, Iterable, Optional, runtime_checkable
from .models import SkillProperties
@@ -45,8 +45,7 @@ class SkillMutatorProtocol(Protocol):
"""Agent-managed skill CRUD operations.
Allows agents to create, edit, and manage their own skills at runtime.
Implementations should provide safe-by-default behaviour: use ``propose=True``
(default) to stage mutations for human approval before writing to disk.
Implementations should provide safe-by-default behavior (e.g., propose mode).
"""
def create(self, name: str, content: str, category: Optional[str] = None,
@@ -164,30 +163,3 @@ class SkillMutatorProtocol(Protocol):
Status message describing the action taken
"""
...
@runtime_checkable
class SkillsCatalogProtocol(Protocol):
"""Metadata listing for skills — UI catalog pages."""
def list_skills(self) -> List[Dict[str, Any]]:
"""Return skill metadata dicts (name, description, location)."""
...
def list_skills_for_api() -> List[Dict[str, Any]]:
"""Default catalog adapter using SkillManager when available."""
try:
from .manager import SkillManager
manager = SkillManager()
return [
{
"name": s.name,
"description": s.description,
"location": getattr(s, "location", ""),
}
for s in manager.get_available_skills()
]
except Exception:
return []
@@ -191,11 +191,9 @@ def validate(skill_dir: Path, strict: bool = False) -> List[str]:
return ["Missing required file: SKILL.md"]
try:
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
metadata, _ = parse_frontmatter(content)
except ParseError as e:
return [str(e)]
except UnicodeDecodeError as e:
return [f"SKILL.md is not valid UTF-8: {e}"]
return validate_metadata(metadata, skill_dir, strict=strict)
@@ -346,15 +346,13 @@ class _LazyStorageModule:
_BACKENDS = {
"RedisBackend": "praisonai.storage",
"MongoDBBackend": "praisonai.storage",
"MongoDBBackend": "praisonai.storage",
"PostgreSQLBackend": "praisonai.storage",
"DynamoDBBackend": "praisonai.storage",
"ValkeyBackend": "praisonai.storage",
"ValkeySearchBackend": "praisonai.storage",
# Legacy names for compatibility
"RedisStorageAdapter": "praisonai.storage",
"MongoDBStorageAdapter": "praisonai.storage",
"PostgreSQLStorageAdapter": "praisonai.storage",
"PostgreSQLStorageAdapter": "praisonai.storage",
"DynamoDBStorageAdapter": "praisonai.storage",
}
@@ -392,7 +390,7 @@ def get_backend(
Factory function to get a storage backend.
Args:
backend_type: Type of backend ("file", "sqlite", "redis", "mongodb", "postgresql", "dynamodb", "valkey")
backend_type: Type of backend ("file", "sqlite", "redis", "mongodb", "postgresql", "dynamodb")
**kwargs: Backend-specific arguments
Returns:
@@ -433,13 +431,9 @@ def get_backend(
# Lazy load from wrapper
backend_class = getattr(_heavy_backends, "DynamoDBBackend")
return backend_class(**kwargs)
elif backend_type == "valkey":
# Lazy load from wrapper
backend_class = getattr(_heavy_backends, "ValkeyBackend")
return backend_class(**kwargs)
else:
raise ValueError(f"Unknown backend type: {backend_type}. "
f"Supported: file, sqlite, redis, mongodb, postgresql, dynamodb, valkey")
f"Supported: file, sqlite, redis, mongodb, postgresql, dynamodb")
def __getattr__(name: str):
"""
@@ -448,8 +442,7 @@ def __getattr__(name: str):
This allows `from praisonaiagents.storage.backends import RedisBackend` to work
while maintaining lazy loading.
"""
if name in {"RedisBackend", "MongoDBBackend", "PostgreSQLBackend", "DynamoDBBackend",
"ValkeyBackend", "ValkeySearchBackend"}:
if name in {"RedisBackend", "MongoDBBackend", "PostgreSQLBackend", "DynamoDBBackend"}:
return getattr(_heavy_backends, name)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -100,107 +100,3 @@ class InMemoryTokenUsageSink:
def clear(self) -> None:
"""Clear all records."""
self.records.clear()
@runtime_checkable
class UsageQueryProtocol(Protocol):
"""Read path for token usage — UI dashboards and analytics."""
def get_summary(self) -> Dict[str, Any]:
"""Aggregate totals and breakdowns by model/agent."""
...
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
"""Recent usage records, newest last."""
...
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
...
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
...
class InMemoryUsageQuery:
"""Query adapter backed by InMemoryTokenUsageSink records."""
def __init__(self, sink: InMemoryTokenUsageSink):
self._sink = sink
def get_summary(self) -> Dict[str, Any]:
records = self._sink.records
total_in = 0
total_out = 0
total_all = 0
by_model: Dict[str, Dict[str, int]] = {}
by_agent: Dict[str, Dict[str, int]] = {}
# Single-pass optimization: calculate everything in one loop
for r in records:
r_in = r.get("input_tokens", 0)
r_out = r.get("output_tokens", 0)
r_total = r.get("total_tokens", r_in + r_out)
total_in += r_in
total_out += r_out
total_all += r_total
model = r.get("model", "unknown")
agent = r.get("agent_name", "unknown")
# Update by_model and by_agent efficiently
for bucket, key in ((by_model, model), (by_agent, agent)):
stats = bucket.setdefault(key, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0})
stats["input_tokens"] += r_in
stats["output_tokens"] += r_out
stats["total_tokens"] += r_total
return {
"total_requests": len(records),
"total_input_tokens": total_in,
"total_output_tokens": total_out,
"total_tokens": total_all, # Use consistent calculation
"by_model": by_model,
"by_agent": by_agent,
}
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
return self._sink.records[-limit:]
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
return self._sink.get_by_task(task_id)
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
return self._sink.get_by_agent(agent_name)
class TokenCollectorUsageQuery:
"""Query adapter reading from the global TokenCollector + optional sink."""
def __init__(self, collector: Any):
self._collector = collector
def get_summary(self) -> Dict[str, Any]:
summary = self._collector.get_session_summary()
return {
"total_requests": summary.get("total_interactions", 0),
"total_input_tokens": summary.get("total_metrics", {}).get("input_tokens", 0),
"total_output_tokens": summary.get("total_metrics", {}).get("output_tokens", 0),
"total_tokens": summary.get("total_tokens", 0),
"by_model": summary.get("by_model", {}),
"by_agent": summary.get("by_agent", {}),
}
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
return self._collector.get_recent_interactions(limit)
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
sink = getattr(self._collector, "_sink", None)
if sink and hasattr(sink, "get_by_task"):
return sink.get_by_task(task_id)
return []
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
sink = getattr(self._collector, "_sink", None)
if sink and hasattr(sink, "get_by_agent"):
return sink.get_by_agent(agent_name)
return []
@@ -191,9 +191,4 @@ class TokenCollector:
# Global token collector instance
_token_collector = TokenCollector()
def get_token_collector() -> TokenCollector:
"""Return the global TokenCollector singleton."""
return _token_collector
_token_collector = TokenCollector()
@@ -7,7 +7,7 @@ import logging
from praisonaiagents._logging import get_logger
# Import dependencies
from .token_collector import get_token_collector, TokenMetrics
from .token_collector import _token_collector, TokenMetrics
from .telemetry import get_telemetry
logger = get_logger(__name__)
@@ -56,12 +56,11 @@ class TokenTelemetryBridge:
def export_token_metrics(self) -> Dict[str, Any]:
"""Export token metrics for telemetry reporting."""
collector = get_token_collector()
if not collector:
if not _token_collector:
return {}
try:
metrics = collector.export_metrics()
metrics = _token_collector.export_metrics()
# Prepare telemetry-friendly format
return {
@@ -78,7 +77,8 @@ class TokenTelemetryBridge:
def reset_token_metrics(self):
"""Reset token metrics collection."""
get_token_collector().reset()
if _token_collector:
_token_collector.reset()
# Global telemetry bridge instance
_token_telemetry_bridge = TokenTelemetryBridge()
@@ -10,7 +10,7 @@ _tools_lazy_cache = {}
# Export core tool items for organized imports (lightweight)
from .base import BaseTool, ToolResult, ToolValidationError, validate_tool
from .decorator import tool, FunctionTool
from .registry import get_registry, register_tool, get_tool, add_tool, has_tool, remove_tool, list_tools, list_available_tools, list_tools_with_allowed_filter, list_tools_with_hermes_filter, ToolRegistry
from .registry import get_registry, register_tool, get_tool, add_tool, has_tool, remove_tool, list_tools, list_available_tools, ToolRegistry
from .tools import Tools
# Export Injected type directly for easy access
@@ -75,12 +75,12 @@ TOOL_MAPPINGS = {
# Python Tools
'execute_code': ('.python_tools', None),
'analyze_code': ('.python_tools', None),
'format_code': ('.python_tools', None),
'lint_code': ('.python_tools', None),
'disassemble_code': ('.python_tools', None),
'python_tools': ('.python_tools', None),
'execute_code': ('.python_tools', 'PythonTools'),
'analyze_code': ('.python_tools', 'PythonTools'),
'format_code': ('.python_tools', 'PythonTools'),
'lint_code': ('.python_tools', 'PythonTools'),
'disassemble_code': ('.python_tools', 'PythonTools'),
'python_tools': ('.python_tools', 'PythonTools'),
# Chain of Thought Training Tools
@@ -326,7 +326,7 @@ def __getattr__(name: str) -> Any:
if name in [
'duckduckgo', 'internet_search', 'searxng_search', 'searxng',
'scrape_page', 'extract_links', 'crawl', 'extract_text',
'execute_command', 'execute_code', 'analyze_code', 'format_code', 'lint_code', 'disassemble_code', 'list_processes', 'kill_process', 'get_system_info',
'execute_command', 'list_processes', 'kill_process', 'get_system_info',
'tavily', 'tavily_search', 'tavily_extract', 'tavily_crawl', 'tavily_map',
'tavily_search_async', 'tavily_extract_async',
'ydc', 'ydc_search', 'ydc_contents', 'ydc_news', 'ydc_images',
@@ -363,7 +363,7 @@ __all__ = list(TOOL_MAPPINGS.keys()) + [
'BaseTool', 'ToolResult', 'ToolValidationError', 'validate_tool',
'tool', 'FunctionTool',
'get_registry', 'register_tool', 'get_tool', 'add_tool', 'has_tool', 'remove_tool',
'list_tools', 'list_available_tools', 'list_tools_with_allowed_filter', 'list_tools_with_hermes_filter', 'ToolRegistry',
'list_tools', 'list_available_tools', 'ToolRegistry',
'Tools',
# Validation and retry protocols
'ValidationResult', 'ToolValidatorProtocol', 'AsyncToolValidatorProtocol', 'PassthroughValidator',
@@ -1,66 +0,0 @@
"""
A2UI tools — optional declarative UI via Google's a2ui-agent-sdk.
Usage:
from praisonaiagents import Agent
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
agent = Agent(name="assistant", tools=[send_a2ui_messages])
"""
from __future__ import annotations
import json
from typing import Any, Dict, List, Union
from praisonaiagents.tools.decorator import tool
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
def _unwrap_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> List[Dict[str, Any]]:
"""Minimal unwrap for common LLM arg shapes (not full UI normalisation)."""
if isinstance(messages, str):
try:
messages = json.loads(messages)
except json.JSONDecodeError as exc:
raise ValueError(f"messages JSON string is not valid JSON: {exc}") from exc
if isinstance(messages, dict):
if "messages" in messages:
messages = messages["messages"]
else:
messages = [messages]
if not isinstance(messages, list):
raise ValueError("messages must be a list of A2UI message dicts or a JSON string")
return messages
@tool
def send_a2ui_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> Dict[str, Any]:
"""
Send validated A2UI JSON messages to a connected A2UI renderer client.
Pass a list of A2UI message dicts (createSurface, updateComponents, etc.)
or a JSON string. Requires: pip install praisonaiagents[a2ui]
Args:
messages: A2UI message list or JSON string from the model.
Returns:
Dict with mime_type and serialised messages for A2A/AG-UI clients.
"""
from praisonaiagents.ui.a2ui.adapter import create_a2ui_part
messages = _unwrap_messages(messages)
part = create_a2ui_part({"messages": messages})
data = getattr(getattr(part, "root", part), "data", messages)
metadata = getattr(getattr(part, "root", part), "metadata", {}) or {}
return {
"mime_type": metadata.get("mimeType", A2UI_MIME_TYPE),
"messages": messages,
"a2ui_part": data,
}
@@ -15,7 +15,7 @@ Usage:
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Type, get_type_hints
from typing import Any, Dict, Optional, Type, get_type_hints
import inspect
import logging
@@ -234,81 +234,18 @@ class BaseTool(ABC):
if getattr(self.run, '__isabstractmethod__', False):
errors.append("Tool must implement the 'run()' method")
# Check parameters schema is valid and OpenAI-compatible
# Check parameters schema is valid
if self.parameters:
if not isinstance(self.parameters, dict):
errors.append("'parameters' must be a dictionary")
elif "type" not in self.parameters:
errors.append("'parameters' must have a 'type' field")
elif "properties" not in self.parameters:
errors.append("'parameters' must have a 'properties' field for OpenAI compatibility")
# Validate schema structure for OpenAI compatibility
try:
schema = self.get_schema()
if not isinstance(schema, dict):
errors.append("get_schema() must return a dictionary")
elif schema.get("type") != "function":
errors.append("get_schema() must return schema with type='function'")
elif "function" not in schema:
errors.append("get_schema() must have 'function' key")
else:
func = schema["function"]
if not isinstance(func.get("parameters"), dict):
errors.append("Schema function.parameters must be a dictionary")
elif "properties" not in func["parameters"]:
errors.append("Schema function.parameters must have 'properties' for OpenAI compatibility")
except Exception as e:
errors.append(f"Schema generation failed: {e}")
if errors:
raise ToolValidationError(f"Tool '{self.name}' validation failed: {'; '.join(errors)}")
return True
def validate_schema_roundtrip(self) -> bool:
"""Validate that schema can round-trip through OpenAI-style serialization.
This ensures the tool schema is compatible with LLM providers that expect
OpenAI function calling format and that parameters can be properly parsed.
Raises:
ToolValidationError: If round-trip fails
Returns:
True if round-trip validation passes
"""
try:
import json
# Get the schema
schema = self.get_schema()
# Serialize to JSON (what gets sent to LLM)
serialized = json.dumps(schema)
# Deserialize back
deserialized = json.loads(serialized)
# Validate structure matches expectations
if deserialized != schema:
raise ToolValidationError(f"Schema serialization round-trip failed for tool '{self.name}'")
# Validate OpenAI format requirements
func_schema = deserialized.get("function", {})
parameters = func_schema.get("parameters", {})
if not isinstance(parameters.get("properties"), dict):
raise ToolValidationError(f"Tool '{self.name}' parameters.properties must be a dict for OpenAI compatibility")
if "required" in parameters and not isinstance(parameters["required"], list):
raise ToolValidationError(f"Tool '{self.name}' parameters.required must be a list")
return True
except Exception as e:
raise ToolValidationError(f"Schema round-trip validation failed for tool '{self.name}': {e}")
@classmethod
def validate_class(cls) -> bool:
"""Validate a tool class before instantiation.
@@ -345,9 +282,7 @@ def validate_tool(tool: Any) -> bool:
ToolValidationError: If validation fails
"""
if isinstance(tool, BaseTool):
tool.validate()
tool.validate_schema_roundtrip()
return True
return tool.validate()
if callable(tool):
# For plain functions, check they have a name
@@ -359,67 +294,5 @@ def validate_tool(tool: Any) -> bool:
raise ToolValidationError(f"Invalid tool type: {type(tool)}")
def validate_tool_schema_consistency(tools: List[Any]) -> bool:
"""Validate a list of tools for schema consistency with OpenAI format.
This function ensures all tools in a list can be properly serialized
and have consistent schema structures for use with LLM providers.
Args:
tools: List of tool objects to validate
Returns:
True if all tools are valid and consistent
Raises:
ToolValidationError: If validation fails
"""
if not tools:
return True
import json
from .decorator import get_tool_schema
schemas = []
for i, tool in enumerate(tools):
try:
# Validate individual tool
validate_tool(tool)
# Get schema from different tool types
if isinstance(tool, BaseTool):
schema = tool.get_schema()
elif hasattr(tool, 'get_schema') and callable(getattr(tool, 'get_schema')):
schema = tool.get_schema()
elif callable(tool):
schema = get_tool_schema(tool)
else:
raise ToolValidationError(f"Cannot extract schema from tool at index {i}: {type(tool)}")
if not schema:
raise ToolValidationError(f"Tool at index {i} returned empty schema")
# Validate JSON serialization
try:
json.dumps(schema)
except (TypeError, ValueError) as e:
raise ToolValidationError(f"Tool schema at index {i} is not JSON serializable: {e}")
schemas.append(schema)
except Exception as e:
raise ToolValidationError(f"Tool validation failed at index {i}: {e}")
# Check for duplicate tool names
names = set()
for schema in schemas:
name = schema.get("function", {}).get("name")
if name in names:
raise ToolValidationError(f"Duplicate tool name '{name}' found in tool list")
names.add(name)
return True
# For backward compatibility - tools can also just be functions
# The @tool decorator (in decorator.py) wraps functions into BaseTool instances
@@ -214,13 +214,6 @@ def tool(
availability=availability
)
# Validate the tool at creation time for early error detection
try:
tool_instance.validate()
tool_instance.validate_schema_roundtrip()
except Exception as e:
logging.warning(f"Tool validation warning for {tool_instance.name}: {e}")
# Register with global registry if available
try:
from .registry import get_registry
@@ -268,11 +268,6 @@ class MentionsParser:
def _process_url_mention(self, url: str) -> Optional[str]:
"""Process @url:https://... mention."""
try:
from praisonaiagents.tools.spider_tools import SpiderTools
if not SpiderTools()._validate_url(url):
return f"# URL: {url}\n[Blocked: URL is not allowed]"
import urllib.request
req = urllib.request.Request(
@@ -36,30 +36,6 @@ def _safe_getattr(obj, name, *default):
return getattr(obj, name, *default) if default else getattr(obj, name)
_SANDBOX_BLOCKED_ATTRS = frozenset({
'__subclasses__', '__bases__', '__mro__', '__globals__',
'__code__', '__class__', '__dict__', '__builtins__',
'__import__', '__loader__', '__spec__', '__init_subclass__',
'__set_name__', '__reduce__', '__reduce_ex__',
'__traceback__', '__qualname__', '__module__',
'__wrapped__', '__closure__', '__annotations__',
'__self__', # C builtins leak real builtins module (GHSA-4mr5-g6f9-cfrh)
# Frame/code object introspection
'gi_frame', 'gi_code', 'cr_frame', 'cr_code',
'ag_frame', 'ag_code', 'tb_frame', 'tb_next',
'f_globals', 'f_locals', 'f_builtins', 'f_code',
'co_consts', 'co_names',
'__getattribute__', '__getattr__', '__setattr__', '__delattr__',
'__dir__', '__get__', '__set__', '__delete__',
})
_SANDBOX_BLOCKED_CALLS = frozenset({
'exec', 'eval', 'compile', '__import__',
'open', 'input', 'breakpoint',
'setattr', 'delattr', 'dir', 'vars',
})
def _validate_code_ast(code: str):
"""Validate code using AST — catches attacks that bypass text checks.
@@ -72,6 +48,23 @@ def _validate_code_ast(code: str):
except SyntaxError:
return None # let compile() handle syntax errors later
# Dangerous dunder attributes attackers use for sandbox escape
_blocked_attrs = frozenset({
'__subclasses__', '__bases__', '__mro__', '__globals__',
'__code__', '__class__', '__dict__', '__builtins__',
'__import__', '__loader__', '__spec__', '__init_subclass__',
'__set_name__', '__reduce__', '__reduce_ex__',
'__traceback__', '__qualname__', '__module__',
'__wrapped__', '__closure__', '__annotations__',
# Frame/code object introspection
'gi_frame', 'gi_code', 'cr_frame', 'cr_code',
'ag_frame', 'ag_code', 'tb_frame', 'tb_next',
'f_globals', 'f_locals', 'f_builtins', 'f_code',
'co_consts', 'co_names',
'__getattribute__', '__getattr__', '__setattr__', '__delattr__',
'__dir__', '__get__', '__set__', '__delete__',
})
for node in ast.walk(tree):
# Block import statements
if isinstance(node, (ast.Import, ast.ImportFrom)):
@@ -79,26 +72,28 @@ def _validate_code_ast(code: str):
# Block attribute access to dangerous dunders
if isinstance(node, ast.Attribute):
if node.attr in _SANDBOX_BLOCKED_ATTRS:
if node.attr in _blocked_attrs:
return (
f"Access to attribute '{node.attr}' is restricted"
)
# Block calls to dangerous builtins (bare name or attribute access)
# Block calls to dangerous builtins by name
if isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Name) and func.id in _SANDBOX_BLOCKED_CALLS:
if isinstance(func, ast.Name) and func.id in (
'exec', 'eval', 'compile', '__import__',
'open', 'input', 'breakpoint',
'setattr', 'delattr', 'dir',
):
return f"Call to '{func.id}' is not allowed"
if isinstance(func, ast.Attribute) and func.attr in _SANDBOX_BLOCKED_CALLS:
return f"Call to '{func.attr}' is not allowed"
# Block dangerous constants (strings containing dunders)
# Fallback for Python 3.7 ast.Str
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if any(attr in node.value for attr in _SANDBOX_BLOCKED_ATTRS):
if any(attr in node.value for attr in _blocked_attrs):
return f"String constant contains restricted attribute name"
elif type(node).__name__ == 'Str':
if any(attr in getattr(node, 's', '') for attr in _SANDBOX_BLOCKED_ATTRS):
if any(attr in getattr(node, 's', '') for attr in _blocked_attrs):
return f"String constant contains restricted attribute name"
return None
@@ -117,16 +112,7 @@ def _execute_code_sandboxed(
"""
if limits is None:
limits = ResourceLimits.minimal()
ast_error = _validate_code_ast(code)
if ast_error:
return {
'result': None,
'stdout': '',
'stderr': f'Security Error: {ast_error}',
'success': False,
}
try:
# Create temporary file for the code
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
@@ -165,8 +151,21 @@ def safe_execute():
}}
# Block dangerous patterns
blocked_attrs = {set(_SANDBOX_BLOCKED_ATTRS)!r}
blocked_calls = {set(_SANDBOX_BLOCKED_CALLS)!r}
blocked_attrs = {{
'__subclasses__', '__bases__', '__mro__', '__globals__',
'__code__', '__class__', '__dict__', '__builtins__',
'__import__', '__loader__', '__spec__', '__init_subclass__',
'__set_name__', '__reduce__', '__reduce_ex__',
'__traceback__', '__qualname__', '__module__',
'__wrapped__', '__closure__', '__annotations__',
# Frame/code object introspection
'gi_frame', 'gi_code', 'cr_frame', 'cr_code',
'ag_frame', 'ag_code', 'tb_frame', 'tb_next',
'f_globals', 'f_locals', 'f_builtins', 'f_code',
'co_consts', 'co_names',
'__getattribute__', '__getattr__', '__setattr__', '__delattr__',
'__dir__', '__get__', '__set__', '__delete__',
}}
for node in ast.walk(tree):
if isinstance(node, (ast.Import, ast.ImportFrom)):
@@ -183,20 +182,14 @@ def safe_execute():
"stderr": f"Access to attribute '{{node.attr}}' is restricted",
"success": False
}}
if isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Name) and func.id in blocked_calls:
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
if node.func.id in ('exec', 'eval', 'compile', '__import__',
'open', 'input', 'breakpoint',
'setattr', 'delattr', 'dir'):
return {{
"result": None,
"stdout": "",
"stderr": f"Call to '{{func.id}}' is not allowed",
"success": False
}}
if isinstance(func, ast.Attribute) and func.attr in blocked_calls:
return {{
"result": None,
"stdout": "",
"stderr": f"Call to '{{func.attr}}' is not allowed",
"stderr": f"Call to '{{node.func.id}}' is not allowed",
"success": False
}}
if isinstance(node, ast.Constant) and isinstance(node.value, str):
@@ -488,7 +481,7 @@ def _execute_code_direct(
'__import__', 'import ', 'from ', 'exec', 'eval',
'compile', 'open(', 'file(', 'input(', 'raw_input',
'__subclasses__', '__bases__', '__globals__', '__code__',
'__class__', '__self__', 'globals(', 'locals(', 'vars('
'__class__', 'globals(', 'locals(', 'vars('
]
code_lower = code.lower()
@@ -184,41 +184,6 @@ class ToolRegistry:
return available
def list_tools_with_allowed_filter(self, context: Optional[Dict[str, Any]] = None) -> List[str]:
"""List tool names filtered by ALLOWED_TOOLS environment variable.
This applies the canonical ALLOWED_TOOLS filter to prevent tool
name collisions in multi-environment agent systems.
Args:
context: Optional context for availability checks
Returns:
List of filtered tool names
"""
try:
# Lazy import to avoid circular dependencies
from ..allowed_tools_filter import filter_tools_with_allowed_tools
# Get all available tool names
all_tools = self.list_tools()
# Apply ALLOWED_TOOLS filter with diagnostics
filtered_tools = filter_tools_with_allowed_tools(all_tools, log_diagnostics=True)
return sorted(filtered_tools)
except ImportError:
logging.warning("ALLOWED_TOOLS filter not available, returning all tools")
return self.list_tools()
except ValueError:
# Preserve strict ALLOWED_TOOLS semantics (e.g., CI unknown tools, empty value)
raise
except Exception as e:
logging.error("Error applying ALLOWED_TOOLS filter: %s", e)
# Fallback to all tools on error
return self.list_tools()
def get_all(self) -> Dict[str, Union[BaseTool, Callable]]:
"""Get all registered tools as a dict. Thread-safe."""
with self._lock:
@@ -416,24 +381,3 @@ def list_available_tools(context: Optional[Dict[str, Any]] = None) -> List[Union
List of available tools
"""
return get_registry().list_available_tools(context)
def list_tools_with_allowed_filter(context: Optional[Dict[str, Any]] = None) -> List[str]:
"""List tool names filtered by ALLOWED_TOOLS from the global registry.
This is the canonical entry point for applying ALLOWED_TOOLS filtering
across the PraisonAI ecosystem.
Args:
context: Optional context for availability checks
Returns:
List of filtered tool names
"""
return get_registry().list_tools_with_allowed_filter(context)
# Backward compatibility alias
def list_tools_with_hermes_filter(context: Optional[Dict[str, Any]] = None) -> List[str]:
"""Backward compatibility alias for list_tools_with_allowed_filter."""
return list_tools_with_allowed_filter(context)
@@ -11,8 +11,6 @@ content = scrape_page("https://example.com")
"""
import logging
import ipaddress
import socket
from typing import List, Dict, Union, Optional, Any
from importlib import util
import json
@@ -22,48 +20,6 @@ import os
import hashlib
import time
def _host_is_blocked(hostname: str) -> bool:
"""Return True when hostname resolves to loopback/private/internal targets."""
if not hostname:
return True
host = hostname.lower().rstrip(".")
if host in ("localhost", "0.0.0.0", "::1") or host.endswith(".localhost"):
return True
if host in ("169.254.169.254", "metadata.google.internal"):
return True
if any(host.endswith(suffix) for suffix in (".local", ".internal", ".localdomain")):
return True
def _ip_blocked(ip: ipaddress._BaseAddress) -> bool:
return bool(
ip.is_private or ip.is_loopback or ip.is_reserved or ip.is_link_local
)
if host.isdigit():
try:
return _ip_blocked(ipaddress.ip_address(int(host)))
except (ValueError, OverflowError):
return True
if host.startswith("0x"):
try:
return _ip_blocked(ipaddress.ip_address(int(host, 16)))
except (ValueError, OverflowError):
return True
try:
return _ip_blocked(ipaddress.ip_address(host))
except ValueError:
pass
try:
return _ip_blocked(ipaddress.ip_address(socket.inet_aton(host)))
except OSError:
pass
return False
class SpiderTools:
"""Tools for web scraping and crawling."""
@@ -103,9 +59,31 @@ class SpiderTools:
if not parsed.hostname:
return False
if _host_is_blocked(parsed.hostname):
# Reject local/internal addresses
hostname = parsed.hostname.lower()
# Block localhost and loopback
if hostname in ['localhost', '127.0.0.1', '0.0.0.0', '::1']:
return False
# Block private IP ranges
import ipaddress
try:
ip = ipaddress.ip_address(hostname)
if ip.is_private or ip.is_reserved or ip.is_loopback or ip.is_link_local:
return False
except ValueError:
# Not an IP address, continue with domain validation
pass
# Block common internal domains
if any(hostname.endswith(domain) for domain in ['.local', '.internal', '.localdomain']):
return False
# Block metadata service endpoints
if hostname in ['169.254.169.254', 'metadata.google.internal']:
return False
return True
except Exception:
@@ -253,7 +253,7 @@ class A2A:
from praisonaiagents.ui.a2a.conversion import (
extract_user_input,
praisonai_to_a2a_message,
create_response_artifact,
create_artifact,
)
# Validate params
@@ -286,8 +286,7 @@ class A2A:
else:
raise RuntimeError("No agent or agents configured")
response_msg = praisonai_to_a2a_message(str(response), task_id=task_id)
history = getattr(self.agent, "chat_history", None) if self.agent else None
artifact = create_response_artifact(str(response), chat_history=history)
artifact = create_artifact(str(response))
self.task_store.add_to_history(task_id, response_msg)
self.task_store.add_artifact(task_id, artifact)
self.task_store.update_status(task_id, TaskState.COMPLETED)
@@ -323,8 +322,7 @@ class A2A:
response_msg = praisonai_to_a2a_message(
str(response), task_id=task.id
)
history = getattr(self.agent, "chat_history", None) if self.agent else None
artifact = create_response_artifact(str(response), chat_history=history)
artifact = create_artifact(str(response))
# Update task store
self.task_store.add_to_history(task.id, response_msg)
@@ -16,7 +16,6 @@ from praisonaiagents.ui.a2a.types import (
DataPart,
Artifact,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
def a2a_to_praisonai_messages(messages: List[Message]) -> List[Dict[str, Any]]:
@@ -108,75 +107,6 @@ def praisonai_to_a2a_message(
)
def find_a2ui_tool_result(
chat_history: Optional[List[Dict[str, Any]]],
) -> Optional[Dict[str, Any]]:
"""Return the last A2UI tool result dict from agent chat history, if any."""
if not chat_history:
return None
for msg in reversed(chat_history):
if msg.get("role") != "tool":
continue
content = msg.get("content")
if content is None:
continue
data: Any = content if isinstance(content, dict) else None
if data is None:
try:
data = json.loads(content)
except (json.JSONDecodeError, TypeError):
continue
if isinstance(data, dict) and data.get("mime_type") == A2UI_MIME_TYPE:
return data
return None
def create_a2ui_artifact(
result: Dict[str, Any],
artifact_id: Optional[str] = None,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Artifact:
"""Create an A2A Artifact with a DataPart for an A2UI tool result."""
art_id = artifact_id or f"art-{uuid.uuid4().hex[:12]}"
messages = result.get("messages") or []
a2ui_data = result.get("a2ui_part")
if not isinstance(a2ui_data, dict):
a2ui_data = {"messages": messages}
elif "messages" not in a2ui_data:
a2ui_data = {**a2ui_data, "messages": messages}
return Artifact(
artifact_id=art_id,
name=name,
description=description,
parts=[
DataPart(
data=a2ui_data,
metadata={"mimeType": result.get("mime_type", A2UI_MIME_TYPE)},
)
],
)
def create_response_artifact(
content: str,
chat_history: Optional[List[Dict[str, Any]]] = None,
artifact_id: Optional[str] = None,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Artifact:
"""Build artifact from response text, preferring A2UI DataPart when present in history."""
a2ui = find_a2ui_tool_result(chat_history)
if a2ui:
return create_a2ui_artifact(
a2ui, artifact_id=artifact_id, name=name, description=description
)
return create_artifact(
content, artifact_id=artifact_id, name=name, description=description
)
def create_artifact(
content: str,
artifact_id: Optional[str] = None,
@@ -1,44 +1,126 @@
"""
A2UI integration via Google's a2ui-agent-sdk (optional).
A2UI (Agent-to-User Interface) Protocol Integration for PraisonAI Agents
Declarative agent UI uses the official SDK — not reimplemented in PraisonAI core.
Install: pip install praisonaiagents[a2ui]
This module provides A2UI protocol support, enabling PraisonAI Agents
to generate rich, interactive UIs declaratively via JSON.
A2UI is Google's open standard that allows agents to "speak UI" by sending
declarative JSON describing UI components, which clients then render natively.
Usage:
from praisonaiagents.ui.a2ui import (
# Message types
CreateSurfaceMessage,
UpdateComponentsMessage,
UpdateDataModelMessage,
DeleteSurfaceMessage,
# Component types
TextComponent,
ImageComponent,
ButtonComponent,
CardComponent,
RowComponent,
ColumnComponent,
# Extension helpers
create_a2ui_part,
is_a2ui_part,
)
"""
from praisonaiagents.ui.a2ui.adapter import (
from praisonaiagents.ui.a2ui.types import (
# Data binding
PathBinding,
resolve_string_or_path,
# Action types
ActionContext,
Action,
# Children types
ChildrenTemplate,
# Component types
TextComponent,
ImageComponent,
IconComponent,
ButtonComponent,
CardComponent,
RowComponent,
ColumnComponent,
ListComponent,
TextFieldComponent,
CheckBoxComponent,
SliderComponent,
DateTimeInputComponent,
# Message types
CreateSurfaceMessage,
UpdateComponentsMessage,
UpdateDataModelMessage,
DeleteSurfaceMessage,
# Data part
A2UIDataPart,
)
from praisonaiagents.ui.a2ui.extension import (
A2UI_MIME_TYPE,
A2UI_EXTENSION_URI,
STANDARD_CATALOG_ID,
create_a2ui_part,
generate_a2ui_system_prompt,
get_schema_manager,
is_a2ui_part,
parse_a2ui_response,
get_a2ui_agent_extension,
AgentExtension,
)
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.agent import A2UIAgent
from praisonaiagents.ui.a2ui.templates import (
ChatTemplate,
ListTemplate,
FormTemplate,
DashboardTemplate,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
__all__ = [
"A2UI",
# Data binding
"PathBinding",
"resolve_string_or_path",
# Action types
"ActionContext",
"Action",
# Children types
"ChildrenTemplate",
# Component types
"TextComponent",
"ImageComponent",
"IconComponent",
"ButtonComponent",
"CardComponent",
"RowComponent",
"ColumnComponent",
"ListComponent",
"TextFieldComponent",
"CheckBoxComponent",
"SliderComponent",
"DateTimeInputComponent",
# Message types
"CreateSurfaceMessage",
"UpdateComponentsMessage",
"UpdateDataModelMessage",
"DeleteSurfaceMessage",
# Data part
"A2UIDataPart",
# Extension
"A2UI_MIME_TYPE",
"A2UIToolResultProtocol",
"A2UI_EXTENSION_URI",
"STANDARD_CATALOG_ID",
"create_a2ui_part",
"generate_a2ui_system_prompt",
"get_schema_manager",
"is_a2ui_part",
"parse_a2ui_response",
"get_a2ui_agent_extension",
"AgentExtension",
# Surface builder
"Surface",
# Agent wrapper
"A2UIAgent",
# Templates
"ChatTemplate",
"ListTemplate",
"FormTemplate",
"DashboardTemplate",
]
class A2UI:
"""
Optional A2UI facade (lazy — imports a2ui-agent-sdk on first use).
Example:
from praisonaiagents.ui import A2UI
part = A2UI.create_part({"createSurface": {...}})
prompt = A2UI.system_prompt("You are a helpful assistant.")
"""
create_part = staticmethod(create_a2ui_part)
is_part = staticmethod(is_a2ui_part)
parse_response = staticmethod(parse_a2ui_response)
schema_manager = staticmethod(get_schema_manager)
system_prompt = staticmethod(generate_a2ui_system_prompt)
@@ -1,89 +0,0 @@
"""
Thin adapter over Google's a2ui-agent-sdk (optional dependency).
Install: pip install praisonaiagents[a2ui]
Spec: https://github.com/google/A2UI
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
_A2UI_INSTALL_HINT = (
"A2UI support requires the optional dependency. "
"Install with: pip install praisonaiagents[a2ui]"
)
def _import(name: str):
try:
import importlib
return importlib.import_module(name)
except ImportError as exc:
raise ImportError(
f"{_A2UI_INSTALL_HINT} "
f"(failed to import '{name}'; "
"if already installed, ensure a2a-sdk>=0.3.0,<1.0.0 and not a2a-sdk 1.x)"
) from exc
def create_a2ui_part(a2ui_data: Dict[str, Any]) -> Any:
"""Wrap A2UI payload for A2A transport (MIME: application/json+a2ui)."""
parts = _import("a2ui.a2a.parts")
return parts.create_a2ui_part(a2ui_data)
def is_a2ui_part(part: Any) -> bool:
"""Return True if an A2A part contains A2UI data."""
parts = _import("a2ui.a2a.parts")
return parts.is_a2ui_part(part)
def parse_a2ui_response(text: str) -> List[Any]:
"""Parse and split LLM text into A2A parts (text + validated A2UI JSON)."""
parser = _import("a2ui.parser.parser")
return parser.parse_response(text)
def get_schema_manager(
version: str = "0.9",
catalogs: Optional[List[Any]] = None,
accepts_inline_catalogs: bool = False,
) -> Any:
"""Create an A2uiSchemaManager for LLM system prompts and validation."""
constants = _import("a2ui.schema.constants")
manager = _import("a2ui.schema.manager")
basic = _import("a2ui.basic_catalog.provider")
version_const = constants.VERSION_0_9 if version == "0.9" else version
catalog_configs = catalogs
if catalog_configs is None:
catalog_configs = [basic.BasicCatalog.get_config(version=version_const)]
return manager.A2uiSchemaManager(
version=version_const,
catalogs=catalog_configs,
accepts_inline_catalogs=accepts_inline_catalogs,
)
def generate_a2ui_system_prompt(
role_description: str,
workflow_description: str = "",
ui_description: str = "",
*,
version: str = "0.9",
include_schema: bool = True,
include_examples: bool = True,
) -> str:
"""Build an LLM system prompt with embedded A2UI schema and examples."""
schema_manager = get_schema_manager(version=version)
return schema_manager.generate_system_prompt(
role_description=role_description,
workflow_description=workflow_description,
ui_description=ui_description,
include_schema=include_schema,
include_examples=include_examples,
)
@@ -0,0 +1,358 @@
"""
A2UI Agent Integration
Provides A2UIAgent wrapper for generating A2UI responses from PraisonAI agents.
"""
import uuid
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.types import (
PathBinding,
A2UIDataPart,
)
from praisonaiagents.ui.a2ui.extension import (
create_a2ui_part,
STANDARD_CATALOG_ID,
)
if TYPE_CHECKING:
from praisonaiagents import Agent
class A2UIAgent:
"""
A2UI wrapper for PraisonAI agents.
Provides methods to generate A2UI responses from agent interactions.
Example:
>>> from praisonaiagents import Agent
>>> from praisonaiagents.ui.a2ui import A2UIAgent
>>>
>>> agent = Agent(name="Assistant", role="Helper", goal="Help users")
>>> a2ui_agent = A2UIAgent(agent=agent)
>>>
>>> # Render a simple text response
>>> messages = a2ui_agent.render_text("Hello, World!")
>>>
>>> # Render a list of items
>>> messages = a2ui_agent.render_list(
... title="Results",
... items=[{"name": "Item 1"}, {"name": "Item 2"}]
... )
"""
def __init__(
self,
agent: "Agent",
surface_id: Optional[str] = None,
catalog_id: str = STANDARD_CATALOG_ID,
):
"""
Initialize A2UIAgent wrapper.
Args:
agent: The PraisonAI Agent to wrap
surface_id: Unique ID for the UI surface (auto-generated if not provided)
catalog_id: The component catalog to use
"""
self.agent = agent
self.surface_id = surface_id or f"surface-{uuid.uuid4().hex[:8]}"
self.catalog_id = catalog_id
self._surface: Optional[Surface] = None
def get_surface(self) -> Surface:
"""
Get the underlying Surface object.
Creates a new Surface if one doesn't exist.
Returns:
The Surface object
"""
if self._surface is None:
self._surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
return self._surface
def _reset_surface(self) -> Surface:
"""Reset and return a new Surface."""
self._surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
return self._surface
# =========================================================================
# Rendering methods
# =========================================================================
def render_text(
self,
text: str,
title: Optional[str] = None,
usage_hint: str = "body",
) -> List[Dict[str, Any]]:
"""
Render a simple text response as A2UI.
Args:
text: The text content to display
title: Optional title heading
usage_hint: Text style hint (h1, h2, body, caption)
Returns:
List of A2UI message dictionaries
"""
surface = self._reset_surface()
children = []
if title:
surface.text(id="title", text=title, usage_hint="h1")
children.append("title")
surface.text(id="content", text=text, usage_hint=usage_hint)
children.append("content")
# Add root column
surface.column(id="root", children=children)
return surface.to_messages()
def render_list(
self,
title: str,
items: List[Dict[str, Any]],
item_title_key: str = "title",
item_description_key: str = "description",
item_image_key: Optional[str] = None,
item_action: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""
Render a list of items as A2UI cards.
Args:
title: List title
items: List of item dictionaries
item_title_key: Key for item title in each dict
item_description_key: Key for item description
item_image_key: Optional key for item image URL
item_action: Optional action name for item buttons
Returns:
List of A2UI message dictionaries
"""
surface = self._reset_surface()
# Title
surface.text(id="list-title", text=title, usage_hint="h1")
# Create card for each item
card_ids = []
for i, item in enumerate(items):
card_id = f"card-{i}"
content_id = f"card-content-{i}"
title_id = f"item-title-{i}"
desc_id = f"item-desc-{i}"
# Item title
surface.text(
id=title_id,
text=item.get(item_title_key, f"Item {i+1}"),
usage_hint="h3"
)
# Item description
surface.text(
id=desc_id,
text=item.get(item_description_key, ""),
usage_hint="body"
)
content_children = [title_id, desc_id]
# Optional action button
if item_action:
btn_id = f"btn-{i}"
btn_text_id = f"btn-text-{i}"
surface.text(id=btn_text_id, text="Select")
surface.button(
id=btn_id,
child=btn_text_id,
action_name=item_action,
action_context=[{"key": "index", "value": i}],
primary=True
)
content_children.append(btn_id)
# Card content column
surface.column(id=content_id, children=content_children)
# Card
surface.card(id=card_id, child=content_id)
card_ids.append(card_id)
# List column
surface.column(id="item-list", children=card_ids)
# Root column
surface.column(id="root", children=["list-title", "item-list"])
# Set data
surface.set_data("items", items)
return surface.to_messages()
def render_form(
self,
title: str,
fields: List[Dict[str, str]],
submit_action: str = "submit",
submit_label: str = "Submit",
) -> List[Dict[str, Any]]:
"""
Render a form as A2UI.
Args:
title: Form title
fields: List of field definitions with id, label, type
submit_action: Action name for submit button
submit_label: Label for submit button
Returns:
List of A2UI message dictionaries
"""
surface = self._reset_surface()
# Title
surface.text(id="form-title", text=title, usage_hint="h1")
# Fields
field_ids = []
form_data = {}
for field in fields:
field_id = field.get("id", f"field-{len(field_ids)}")
label = field.get("label", field_id)
field_type = field.get("type", "text")
surface.text_field(
id=field_id,
label=label,
text=PathBinding(path=f"/{field_id}"),
field_type=field_type,
)
field_ids.append(field_id)
form_data[field_id] = field.get("default", "")
# Submit button
surface.text(id="submit-text", text=submit_label)
# Build action context from fields
action_context = [
{"key": f["id"], "value": {"path": f"/{f['id']}"}}
for f in fields
]
surface.button(
id="submit-btn",
child="submit-text",
action_name=submit_action,
action_context=action_context,
primary=True,
)
field_ids.append("submit-btn")
# Form column
surface.column(id="form-fields", children=field_ids)
# Root column
surface.column(id="root", children=["form-title", "form-fields"])
# Set initial form data
for key, value in form_data.items():
surface.set_data(key, value)
return surface.to_messages()
# =========================================================================
# Agent interaction methods
# =========================================================================
async def chat_async(self, message: str) -> str:
"""
Send a message to the agent and get a response.
Args:
message: The message to send
Returns:
The agent's response text
"""
# Use the agent's chat method
if hasattr(self.agent, 'chat'):
response = self.agent.chat(message)
return response
elif hasattr(self.agent, 'run'):
response = self.agent.run(message)
return response
else:
raise NotImplementedError("Agent does not have chat or run method")
def chat(self, message: str) -> str:
"""
Send a message to the agent and get a response (sync).
Args:
message: The message to send
Returns:
The agent's response text
"""
if hasattr(self.agent, 'chat'):
return self.agent.chat(message)
elif hasattr(self.agent, 'run'):
return self.agent.run(message)
else:
raise NotImplementedError("Agent does not have chat or run method")
# =========================================================================
# Export methods
# =========================================================================
def to_messages(self) -> List[Dict[str, Any]]:
"""
Get the current surface as A2UI messages.
Returns:
List of A2UI message dictionaries
"""
return self.get_surface().to_messages()
def to_json(self, indent: Optional[int] = None) -> str:
"""
Get the current surface as JSON string.
Args:
indent: JSON indentation level
Returns:
JSON string of A2UI messages
"""
return self.get_surface().to_json(indent=indent)
def to_a2a_part(self) -> A2UIDataPart:
"""
Get the current surface as an A2A DataPart.
Returns:
A2UIDataPart with A2UI MIME type
"""
messages = self.to_messages()
# For A2A, we send messages as a list
return create_a2ui_part({"messages": messages})
@@ -0,0 +1,115 @@
"""
A2UI Extension for A2A Protocol
Provides helpers for integrating A2UI with the A2A (Agent-to-Agent) protocol.
"""
from dataclasses import dataclass
from typing import Any, Dict, Optional
from praisonaiagents.ui.a2ui.types import A2UIDataPart
# =============================================================================
# Constants
# =============================================================================
A2UI_EXTENSION_URI = "https://a2ui.org/a2a-extension/a2ui/v0.9"
A2UI_MIME_TYPE = "application/json+a2ui"
STANDARD_CATALOG_ID = "https://raw.githubusercontent.com/google/A2UI/refs/heads/main/specification/0.9/json/standard_catalog_definition.json"
# =============================================================================
# Agent Extension Type
# =============================================================================
@dataclass
class AgentExtension:
"""
A2A Agent Extension configuration.
Describes an extension that an agent supports.
"""
uri: str
description: str = ""
params: Optional[Dict[str, Any]] = None
# =============================================================================
# Helper Functions
# =============================================================================
def create_a2ui_part(a2ui_data: Dict[str, Any]) -> A2UIDataPart:
"""
Create an A2UI data part for A2A integration.
Wraps A2UI data in an A2UIDataPart with the appropriate MIME type.
Args:
a2ui_data: The A2UI data dictionary (e.g., createSurface, updateComponents)
Returns:
An A2UIDataPart with the A2UI MIME type in metadata
Example:
>>> part = create_a2ui_part({"createSurface": {"surfaceId": "main", "catalogId": "..."}})
>>> part.metadata["mimeType"]
'application/json+a2ui'
"""
return A2UIDataPart(
data=a2ui_data,
metadata={"mimeType": A2UI_MIME_TYPE}
)
def is_a2ui_part(part: A2UIDataPart) -> bool:
"""
Check if a data part contains A2UI data.
Args:
part: The data part to check
Returns:
True if the part contains A2UI data (has A2UI MIME type), False otherwise
Example:
>>> part = create_a2ui_part({"createSurface": {...}})
>>> is_a2ui_part(part)
True
"""
if not isinstance(part, A2UIDataPart):
return False
if part.metadata is None:
return False
return part.metadata.get("mimeType") == A2UI_MIME_TYPE
def get_a2ui_agent_extension(
accepts_inline_custom_catalog: bool = False,
) -> AgentExtension:
"""
Create the A2UI AgentExtension configuration.
Returns an AgentExtension that can be added to an A2A Agent Card
to indicate A2UI support.
Args:
accepts_inline_custom_catalog: Whether the agent accepts inline custom catalogs
Returns:
The configured A2UI AgentExtension
Example:
>>> ext = get_a2ui_agent_extension()
>>> ext.uri
'https://a2ui.org/a2a-extension/a2ui/v0.9'
"""
params: Optional[Dict[str, Any]] = None
if accepts_inline_custom_catalog:
params = {"acceptsInlineCustomCatalog": True}
return AgentExtension(
uri=A2UI_EXTENSION_URI,
description="Provides agent-driven UI using the A2UI JSON format.",
params=params
)
@@ -0,0 +1,364 @@
"""
A2UI Surface Builder
Provides a fluent API for building A2UI surfaces with components and data.
"""
import json
from typing import Any, Dict, List, Optional
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
from praisonaiagents.ui.a2ui.types import (
Component,
TextComponent,
ImageComponent,
ButtonComponent,
CardComponent,
RowComponent,
ColumnComponent,
ListComponent,
TextFieldComponent,
Action,
ActionContext,
PathBinding,
CreateSurfaceMessage,
UpdateComponentsMessage,
UpdateDataModelMessage,
StringOrPath,
ChildrenType,
)
class Surface:
"""
A2UI Surface builder for creating rich agent-generated UIs.
Provides a fluent API for adding components and data, then
generating the A2UI JSON messages.
Example:
>>> surface = Surface(surface_id="main")
>>> surface.text(id="title", text="Hello World", usage_hint="h1")
>>> surface.button(id="submit", child="submit-text", action_name="submit")
>>> surface.set_data("title", "Hello World")
>>> messages = surface.to_messages()
"""
def __init__(
self,
surface_id: str,
catalog_id: str = STANDARD_CATALOG_ID,
):
"""
Initialize a new Surface.
Args:
surface_id: Unique identifier for this surface
catalog_id: The component catalog to use (defaults to standard catalog)
"""
self.surface_id = surface_id
self.catalog_id = catalog_id
self.components: List[Component] = []
self.data: Dict[str, Any] = {}
def add(self, component: Component) -> "Surface":
"""
Add a component to the surface.
Args:
component: The component to add
Returns:
self for method chaining
"""
self.components.append(component)
return self
def set_data(self, key: str, value: Any) -> "Surface":
"""
Set a value in the data model.
Args:
key: The key in the data model
value: The value to set
Returns:
self for method chaining
"""
self.data[key] = value
return self
# =========================================================================
# Helper methods for common components
# =========================================================================
def text(
self,
id: str,
text: StringOrPath,
usage_hint: Optional[str] = None,
weight: Optional[float] = None,
) -> "Surface":
"""
Add a Text component.
Args:
id: Component ID
text: Text content or path binding
usage_hint: Style hint (h1, h2, h3, body, caption)
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(TextComponent(
id=id,
text=text,
usage_hint=usage_hint,
weight=weight,
))
def image(
self,
id: str,
url: StringOrPath,
fit: Optional[str] = None,
usage_hint: Optional[str] = None,
weight: Optional[float] = None,
) -> "Surface":
"""
Add an Image component.
Args:
id: Component ID
url: Image URL or path binding
fit: Image fit (contain, cover, fill, none, scale-down)
usage_hint: Size hint (icon, avatar, smallFeature, etc.)
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(ImageComponent(
id=id,
url=url,
fit=fit,
usage_hint=usage_hint,
weight=weight,
))
def button(
self,
id: str,
child: str,
action_name: Optional[str] = None,
action_context: Optional[List[Dict[str, Any]]] = None,
primary: bool = False,
weight: Optional[float] = None,
) -> "Surface":
"""
Add a Button component.
Args:
id: Component ID
child: Child component ID (usually a Text component)
action_name: Name of the action to trigger
action_context: List of {key, value} dicts for action context
primary: Whether this is a primary button
weight: Flex weight
Returns:
self for method chaining
"""
action = None
if action_name:
context = []
if action_context:
for ctx in action_context:
value = ctx.get("value")
if isinstance(value, dict) and "path" in value:
value = PathBinding(path=value["path"])
context.append(ActionContext(key=ctx["key"], value=value))
action = Action(name=action_name, context=context)
return self.add(ButtonComponent(
id=id,
child=child,
action=action,
primary=primary,
weight=weight,
))
def card(
self,
id: str,
child: str,
weight: Optional[float] = None,
) -> "Surface":
"""
Add a Card component.
Args:
id: Component ID
child: Child component ID
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(CardComponent(
id=id,
child=child,
weight=weight,
))
def row(
self,
id: str,
children: ChildrenType,
weight: Optional[float] = None,
) -> "Surface":
"""
Add a Row (horizontal layout) component.
Args:
id: Component ID
children: List of child IDs or ChildrenTemplate
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(RowComponent(
id=id,
children=children,
weight=weight,
))
def column(
self,
id: str,
children: ChildrenType,
weight: Optional[float] = None,
) -> "Surface":
"""
Add a Column (vertical layout) component.
Args:
id: Component ID
children: List of child IDs or ChildrenTemplate
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(ColumnComponent(
id=id,
children=children,
weight=weight,
))
def list(
self,
id: str,
children: ChildrenType,
direction: str = "vertical",
weight: Optional[float] = None,
) -> "Surface":
"""
Add a List component.
Args:
id: Component ID
children: List of child IDs or ChildrenTemplate for dynamic lists
direction: "vertical" or "horizontal"
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(ListComponent(
id=id,
children=children,
direction=direction,
weight=weight,
))
def text_field(
self,
id: str,
label: StringOrPath,
text: StringOrPath,
field_type: str = "text",
weight: Optional[float] = None,
) -> "Surface":
"""
Add a TextField input component.
Args:
id: Component ID
label: Field label
text: Text value or path binding
field_type: Input type (text, number, email, etc.)
weight: Flex weight
Returns:
self for method chaining
"""
return self.add(TextFieldComponent(
id=id,
label=label,
text=text,
field_type=field_type,
weight=weight,
))
# =========================================================================
# Message generation
# =========================================================================
def to_messages(self) -> List[Dict[str, Any]]:
"""
Generate A2UI messages for this surface.
Returns:
List of A2UI message dictionaries
"""
messages: List[Dict[str, Any]] = []
# 1. CreateSurface message
create_msg = CreateSurfaceMessage(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
messages.append(create_msg.to_dict())
# 2. UpdateComponents message (if we have components)
if self.components:
update_msg = UpdateComponentsMessage(
surface_id=self.surface_id,
components=self.components,
)
messages.append(update_msg.to_dict())
# 3. UpdateDataModel message (if we have data)
if self.data:
data_msg = UpdateDataModelMessage(
surface_id=self.surface_id,
value=self.data,
)
messages.append(data_msg.to_dict())
return messages
def to_json(self, indent: Optional[int] = None) -> str:
"""
Generate A2UI messages as a JSON string.
Args:
indent: JSON indentation level (None for compact)
Returns:
JSON string of A2UI messages
"""
return json.dumps(self.to_messages(), indent=indent)
@@ -0,0 +1,17 @@
"""
A2UI Templates
Pre-built UI templates for common use cases.
"""
from praisonaiagents.ui.a2ui.templates.chat import ChatTemplate
from praisonaiagents.ui.a2ui.templates.list import ListTemplate
from praisonaiagents.ui.a2ui.templates.form import FormTemplate
from praisonaiagents.ui.a2ui.templates.dashboard import DashboardTemplate
__all__ = [
"ChatTemplate",
"ListTemplate",
"FormTemplate",
"DashboardTemplate",
]
@@ -0,0 +1,122 @@
"""
Chat Template for A2UI
Provides a conversational chat UI template.
"""
from typing import Any, Dict, List, Optional
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
class ChatTemplate:
"""
Chat UI template for conversational interfaces.
Creates a chat-style UI with user and agent messages.
Example:
>>> template = ChatTemplate(surface_id="chat")
>>> template.add_user_message("Hello!")
>>> template.add_agent_message("Hi there! How can I help?")
>>> messages = template.to_messages()
"""
def __init__(
self,
surface_id: str = "chat",
catalog_id: str = STANDARD_CATALOG_ID,
title: Optional[str] = None,
):
"""
Initialize ChatTemplate.
Args:
surface_id: Unique ID for the surface
catalog_id: Component catalog to use
title: Optional title for the chat
"""
self.surface_id = surface_id
self.catalog_id = catalog_id
self.title = title
self.messages: List[Dict[str, Any]] = []
def add_user_message(self, content: str) -> "ChatTemplate":
"""
Add a user message.
Args:
content: Message content
Returns:
self for chaining
"""
self.messages.append({
"role": "user",
"content": content,
})
return self
def add_agent_message(self, content: str) -> "ChatTemplate":
"""
Add an agent message.
Args:
content: Message content
Returns:
self for chaining
"""
self.messages.append({
"role": "agent",
"content": content,
})
return self
def to_messages(self) -> List[Dict[str, Any]]:
"""
Generate A2UI messages for the chat.
Returns:
List of A2UI message dictionaries
"""
surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
children = []
# Title if provided
if self.title:
surface.text(id="chat-title", text=self.title, usage_hint="h1")
children.append("chat-title")
# Message components
msg_ids = []
for i, msg in enumerate(self.messages):
msg_id = f"msg-{i}"
role = msg["role"]
content = msg["content"]
# Style based on role
if role == "user":
surface.text(id=msg_id, text=f"You: {content}", usage_hint="body")
else:
surface.text(id=msg_id, text=f"Agent: {content}", usage_hint="body")
msg_ids.append(msg_id)
# Messages column
if msg_ids:
surface.column(id="messages", children=msg_ids)
children.append("messages")
# Root column
surface.column(id="root", children=children)
# Set data
surface.set_data("messages", self.messages)
return surface.to_messages()
@@ -0,0 +1,128 @@
"""
Dashboard Template for A2UI
Provides a dashboard UI template with multiple panels.
"""
from typing import Any, Dict, List, Optional
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
class DashboardTemplate:
"""
Dashboard UI template with multiple panels.
Creates a dashboard layout with cards for each panel.
Example:
>>> template = DashboardTemplate(surface_id="dashboard", title="My Dashboard")
>>> template.add_panel(id="stats", title="Statistics", content="100 items")
>>> template.add_panel(id="chart", title="Chart", content="[Chart here]")
>>> messages = template.to_messages()
"""
def __init__(
self,
surface_id: str = "dashboard",
title: str = "Dashboard",
catalog_id: str = STANDARD_CATALOG_ID,
columns: int = 2,
):
"""
Initialize DashboardTemplate.
Args:
surface_id: Unique ID for the surface
title: Dashboard title
catalog_id: Component catalog to use
columns: Number of columns in the grid
"""
self.surface_id = surface_id
self.title = title
self.catalog_id = catalog_id
self.columns = columns
self.panels: List[Dict[str, Any]] = []
def add_panel(
self,
id: str,
title: str,
content: str,
weight: Optional[float] = None,
) -> "DashboardTemplate":
"""
Add a panel to the dashboard.
Args:
id: Panel ID
title: Panel title
content: Panel content text
weight: Optional flex weight
Returns:
self for chaining
"""
self.panels.append({
"id": id,
"title": title,
"content": content,
"weight": weight,
})
return self
def to_messages(self) -> List[Dict[str, Any]]:
"""
Generate A2UI messages for the dashboard.
Returns:
List of A2UI message dictionaries
"""
surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
# Title
surface.text(id="dashboard-title", text=self.title, usage_hint="h1")
# Create cards for each panel
card_ids = []
for panel in self.panels:
panel_id = panel["id"]
content_id = f"{panel_id}-content"
title_id = f"{panel_id}-title"
text_id = f"{panel_id}-text"
# Panel title
surface.text(id=title_id, text=panel["title"], usage_hint="h3")
# Panel content
surface.text(id=text_id, text=panel["content"], usage_hint="body")
# Panel content column
surface.column(id=content_id, children=[title_id, text_id])
# Panel card
surface.card(id=panel_id, child=content_id, weight=panel.get("weight"))
card_ids.append(panel_id)
# Arrange panels in rows based on columns setting
row_ids = []
for i in range(0, len(card_ids), self.columns):
row_cards = card_ids[i:i + self.columns]
row_id = f"row-{i // self.columns}"
surface.row(id=row_id, children=row_cards)
row_ids.append(row_id)
# Panels container
surface.column(id="panels", children=row_ids)
# Root column
surface.column(id="root", children=["dashboard-title", "panels"])
# Set data
surface.set_data("panels", self.panels)
return surface.to_messages()
@@ -0,0 +1,205 @@
"""
Form Template for A2UI
Provides a form UI template for data collection.
"""
from typing import Any, Dict, List
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.types import PathBinding
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
class FormTemplate:
"""
Form UI template for data collection.
Creates a form with various input fields and a submit button.
Example:
>>> template = FormTemplate(surface_id="form", title="Contact")
>>> template.add_text_field(id="name", label="Name")
>>> template.add_email_field(id="email", label="Email")
>>> template.set_submit_action("submit_form", "Send")
>>> messages = template.to_messages()
"""
def __init__(
self,
surface_id: str = "form",
title: str = "Form",
catalog_id: str = STANDARD_CATALOG_ID,
):
"""
Initialize FormTemplate.
Args:
surface_id: Unique ID for the surface
title: Form title
catalog_id: Component catalog to use
"""
self.surface_id = surface_id
self.title = title
self.catalog_id = catalog_id
self.fields: List[Dict[str, Any]] = []
self.submit_action = "submit"
self.submit_label = "Submit"
def add_text_field(
self,
id: str,
label: str,
default: str = "",
required: bool = False,
) -> "FormTemplate":
"""
Add a text input field.
Args:
id: Field ID
label: Field label
default: Default value
required: Whether field is required
Returns:
self for chaining
"""
self.fields.append({
"id": id,
"label": label,
"type": "text",
"default": default,
"required": required,
})
return self
def add_number_field(
self,
id: str,
label: str,
default: float = 0,
required: bool = False,
) -> "FormTemplate":
"""
Add a number input field.
Args:
id: Field ID
label: Field label
default: Default value
required: Whether field is required
Returns:
self for chaining
"""
self.fields.append({
"id": id,
"label": label,
"type": "number",
"default": default,
"required": required,
})
return self
def add_email_field(
self,
id: str,
label: str,
default: str = "",
required: bool = False,
) -> "FormTemplate":
"""
Add an email input field.
Args:
id: Field ID
label: Field label
default: Default value
required: Whether field is required
Returns:
self for chaining
"""
self.fields.append({
"id": id,
"label": label,
"type": "email",
"default": default,
"required": required,
})
return self
def set_submit_action(
self,
action: str,
label: str = "Submit",
) -> "FormTemplate":
"""
Set the submit button action.
Args:
action: Action name
label: Button label
Returns:
self for chaining
"""
self.submit_action = action
self.submit_label = label
return self
def to_messages(self) -> List[Dict[str, Any]]:
"""
Generate A2UI messages for the form.
Returns:
List of A2UI message dictionaries
"""
surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
# Title
surface.text(id="form-title", text=self.title, usage_hint="h1")
# Fields
field_ids = []
for field in self.fields:
field_id = field["id"]
surface.text_field(
id=field_id,
label=field["label"],
text=PathBinding(path=f"/{field_id}"),
field_type=field["type"],
)
field_ids.append(field_id)
# Set default value
surface.set_data(field_id, field.get("default", ""))
# Submit button
surface.text(id="submit-text", text=self.submit_label)
action_context = [
{"key": f["id"], "value": {"path": f"/{f['id']}"}}
for f in self.fields
]
surface.button(
id="submit-btn",
child="submit-text",
action_name=self.submit_action,
action_context=action_context,
primary=True,
)
field_ids.append("submit-btn")
# Form fields column
surface.column(id="fields", children=field_ids)
# Root column
surface.column(id="root", children=["form-title", "fields"])
return surface.to_messages()
@@ -0,0 +1,147 @@
"""
List Template for A2UI
Provides a list UI template for displaying items.
"""
from typing import Any, Dict, List, Optional
from praisonaiagents.ui.a2ui.surface import Surface
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
class ListTemplate:
"""
List UI template for displaying items.
Creates a list of cards with title, description, and optional image.
Example:
>>> template = ListTemplate(surface_id="list", title="Results")
>>> template.add_item(title="Item 1", description="First item")
>>> template.add_item(title="Item 2", description="Second item")
>>> messages = template.to_messages()
"""
def __init__(
self,
surface_id: str = "list",
title: str = "Items",
catalog_id: str = STANDARD_CATALOG_ID,
):
"""
Initialize ListTemplate.
Args:
surface_id: Unique ID for the surface
title: List title
catalog_id: Component catalog to use
"""
self.surface_id = surface_id
self.title = title
self.catalog_id = catalog_id
self.items: List[Dict[str, Any]] = []
def add_item(
self,
title: str,
description: str = "",
image_url: Optional[str] = None,
action: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> "ListTemplate":
"""
Add an item to the list.
Args:
title: Item title
description: Item description
image_url: Optional image URL
action: Optional action name for item button
metadata: Optional additional metadata
Returns:
self for chaining
"""
item = {
"title": title,
"description": description,
}
if image_url:
item["image_url"] = image_url
if action:
item["action"] = action
if metadata:
item.update(metadata)
self.items.append(item)
return self
def to_messages(self) -> List[Dict[str, Any]]:
"""
Generate A2UI messages for the list.
Returns:
List of A2UI message dictionaries
"""
surface = Surface(
surface_id=self.surface_id,
catalog_id=self.catalog_id,
)
# Title
surface.text(id="list-title", text=self.title, usage_hint="h1")
# Create cards for each item
card_ids = []
for i, item in enumerate(self.items):
card_id = f"card-{i}"
content_id = f"content-{i}"
title_id = f"title-{i}"
desc_id = f"desc-{i}"
# Item title
surface.text(id=title_id, text=item["title"], usage_hint="h3")
# Item description
surface.text(id=desc_id, text=item.get("description", ""), usage_hint="body")
content_children = [title_id, desc_id]
# Optional image
if item.get("image_url"):
img_id = f"img-{i}"
surface.image(id=img_id, url=item["image_url"], usage_hint="smallFeature")
content_children.insert(0, img_id)
# Optional action button
if item.get("action"):
btn_id = f"btn-{i}"
btn_text_id = f"btn-text-{i}"
surface.text(id=btn_text_id, text="Select")
surface.button(
id=btn_id,
child=btn_text_id,
action_name=item["action"],
action_context=[{"key": "index", "value": i}],
primary=True,
)
content_children.append(btn_id)
# Card content
surface.column(id=content_id, children=content_children)
# Card
surface.card(id=card_id, child=content_id)
card_ids.append(card_id)
# Items column
surface.column(id="items", children=card_ids)
# Root column
surface.column(id="root", children=["list-title", "items"])
# Set data
surface.set_data("items", self.items)
return surface.to_messages()
@@ -0,0 +1,523 @@
"""
A2UI Protocol Types
Pydantic-style classes for the A2UI (Agent-to-User Interface) protocol.
Based on the A2UI Specification v0.9.
"""
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
from enum import Enum
# =============================================================================
# Data Binding Types
# =============================================================================
@dataclass
class PathBinding:
"""
A reference to a value in the data model.
Used for data binding in A2UI components.
"""
path: str
def to_dict(self) -> Dict[str, str]:
return {"path": self.path}
# Type alias for values that can be literal or path-bound
StringOrPath = Union[str, PathBinding]
NumberOrPath = Union[int, float, PathBinding]
BooleanOrPath = Union[bool, PathBinding]
def resolve_string_or_path(value: StringOrPath) -> Union[str, Dict[str, str]]:
"""Resolve a StringOrPath to its serialized form."""
if isinstance(value, PathBinding):
return value.to_dict()
return value
def resolve_number_or_path(value: NumberOrPath) -> Union[int, float, Dict[str, str]]:
"""Resolve a NumberOrPath to its serialized form."""
if isinstance(value, PathBinding):
return value.to_dict()
return value
def resolve_boolean_or_path(value: BooleanOrPath) -> Union[bool, Dict[str, str]]:
"""Resolve a BooleanOrPath to its serialized form."""
if isinstance(value, PathBinding):
return value.to_dict()
return value
# =============================================================================
# Action Types
# =============================================================================
@dataclass
class ActionContext:
"""
A key-value pair for action context.
Used to pass data from UI to action handlers.
"""
key: str
value: Union[str, int, float, bool, PathBinding]
def to_dict(self) -> Dict[str, Any]:
if isinstance(self.value, PathBinding):
return {"key": self.key, "value": self.value.to_dict()}
return {"key": self.key, "value": self.value}
@dataclass
class Action:
"""
An action triggered by a UI component (e.g., button click).
"""
name: str
context: List[ActionContext] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"context": [c.to_dict() for c in self.context]
}
# =============================================================================
# Children Types
# =============================================================================
@dataclass
class ChildrenTemplate:
"""
A template for generating dynamic children from a data model list.
"""
component_id: str
path: str
def to_dict(self) -> Dict[str, str]:
return {
"componentId": self.component_id,
"path": self.path
}
# Type alias for children - either explicit list or template
ChildrenType = Union[List[str], ChildrenTemplate]
def resolve_children(children: ChildrenType) -> Union[List[str], Dict[str, str]]:
"""Resolve children to serialized form."""
if isinstance(children, ChildrenTemplate):
return children.to_dict()
return children
# =============================================================================
# Component Types
# =============================================================================
class UsageHint(str, Enum):
"""Text usage hints for styling."""
H1 = "h1"
H2 = "h2"
H3 = "h3"
H4 = "h4"
H5 = "h5"
CAPTION = "caption"
BODY = "body"
class ImageFit(str, Enum):
"""Image fit options."""
CONTAIN = "contain"
COVER = "cover"
FILL = "fill"
NONE = "none"
SCALE_DOWN = "scale-down"
class ImageUsageHint(str, Enum):
"""Image usage hints."""
ICON = "icon"
AVATAR = "avatar"
SMALL_FEATURE = "smallFeature"
MEDIUM_FEATURE = "mediumFeature"
LARGE_FEATURE = "largeFeature"
HEADER = "header"
@dataclass
class TextComponent:
"""Text display component."""
id: str
text: StringOrPath
usage_hint: Optional[str] = None
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Text",
"text": resolve_string_or_path(self.text)
}
if self.usage_hint:
result["usageHint"] = self.usage_hint
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class ImageComponent:
"""Image display component."""
id: str
url: StringOrPath
fit: Optional[str] = None
usage_hint: Optional[str] = None
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Image",
"url": resolve_string_or_path(self.url)
}
if self.fit:
result["fit"] = self.fit
if self.usage_hint:
result["usageHint"] = self.usage_hint
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class IconComponent:
"""Icon display component."""
id: str
name: StringOrPath
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Icon",
"name": resolve_string_or_path(self.name)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class ButtonComponent:
"""Button component with action."""
id: str
child: str
action: Optional[Action] = None
primary: bool = False
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Button",
"child": self.child
}
if self.primary:
result["primary"] = True
if self.action:
result["action"] = self.action.to_dict()
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class CardComponent:
"""Card container component."""
id: str
child: str
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Card",
"child": self.child
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class RowComponent:
"""Horizontal layout component."""
id: str
children: ChildrenType
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Row",
"children": resolve_children(self.children)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class ColumnComponent:
"""Vertical layout component."""
id: str
children: ChildrenType
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Column",
"children": resolve_children(self.children)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class ListComponent:
"""List component for repeating items."""
id: str
children: ChildrenType
direction: str = "vertical"
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "List",
"direction": self.direction,
"children": resolve_children(self.children)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class TextFieldComponent:
"""Text input field component."""
id: str
label: StringOrPath
text: StringOrPath
field_type: str = "text"
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "TextField",
"label": resolve_string_or_path(self.label),
"text": resolve_string_or_path(self.text),
"type": self.field_type
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class CheckBoxComponent:
"""Checkbox input component."""
id: str
label: StringOrPath
checked: BooleanOrPath = False
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "CheckBox",
"label": resolve_string_or_path(self.label),
"checked": resolve_boolean_or_path(self.checked)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class SliderComponent:
"""Slider input component."""
id: str
min_value: NumberOrPath
max_value: NumberOrPath
value: NumberOrPath
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "Slider",
"min": resolve_number_or_path(self.min_value),
"max": resolve_number_or_path(self.max_value),
"value": resolve_number_or_path(self.value)
}
if self.weight is not None:
result["weight"] = self.weight
return result
@dataclass
class DateTimeInputComponent:
"""Date/time input component."""
id: str
label: StringOrPath
value: StringOrPath
enable_date: bool = True
enable_time: bool = False
weight: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"id": self.id,
"component": "DateTimeInput",
"label": resolve_string_or_path(self.label),
"value": resolve_string_or_path(self.value),
"enableDate": self.enable_date,
"enableTime": self.enable_time
}
if self.weight is not None:
result["weight"] = self.weight
return result
# Type alias for any component
Component = Union[
TextComponent,
ImageComponent,
IconComponent,
ButtonComponent,
CardComponent,
RowComponent,
ColumnComponent,
ListComponent,
TextFieldComponent,
CheckBoxComponent,
SliderComponent,
DateTimeInputComponent,
]
# =============================================================================
# Message Types
# =============================================================================
@dataclass
class CreateSurfaceMessage:
"""
Message to create a new UI surface.
Signals the client to create a new surface and begin rendering it.
"""
surface_id: str
catalog_id: str
def to_dict(self) -> Dict[str, Any]:
return {
"createSurface": {
"surfaceId": self.surface_id,
"catalogId": self.catalog_id
}
}
@dataclass
class UpdateComponentsMessage:
"""
Message to update surface components.
Updates a surface with a new set of components.
"""
surface_id: str
components: List[Component]
def to_dict(self) -> Dict[str, Any]:
return {
"updateComponents": {
"surfaceId": self.surface_id,
"components": [c.to_dict() for c in self.components]
}
}
@dataclass
class UpdateDataModelMessage:
"""
Message to update the data model.
Updates the data model for an existing surface.
"""
surface_id: str
value: Any
path: str = "/"
op: str = "replace"
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {
"updateDataModel": {
"surfaceId": self.surface_id,
"path": self.path,
"op": self.op,
"value": self.value
}
}
return result
@dataclass
class DeleteSurfaceMessage:
"""
Message to delete a surface.
Signals the client to delete the surface.
"""
surface_id: str
def to_dict(self) -> Dict[str, Any]:
return {
"deleteSurface": {
"surfaceId": self.surface_id
}
}
# =============================================================================
# Data Part Type
# =============================================================================
@dataclass
class A2UIDataPart:
"""
A2UI data part for A2A integration.
Wraps A2UI data with the appropriate MIME type.
"""
data: Dict[str, Any]
metadata: Optional[Dict[str, Any]] = None
def to_dict(self) -> Dict[str, Any]:
result: Dict[str, Any] = {"data": self.data}
if self.metadata:
result["metadata"] = self.metadata
return result
@@ -4,15 +4,12 @@ AG-UI Event Streaming
Provides utilities for streaming PraisonAI agent responses as AG-UI events.
"""
import asyncio
import json
import uuid
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Set
from praisonaiagents.ui.agui.types import (
BaseEvent,
CustomEvent,
TextMessageStartEvent,
TextMessageContentEvent,
TextMessageEndEvent,
@@ -27,7 +24,6 @@ from praisonaiagents.ui.agui.types import (
StepFinishedEvent,
StateSnapshotEvent,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
@dataclass
@@ -260,75 +256,6 @@ async def async_stream_response(
yield create_run_error_event(str(e))
def stream_event_to_agui_events(
event: Any,
message_id: str,
buffer: EventBuffer,
) -> List[BaseEvent]:
"""Convert a PraisonAI StreamEvent into zero or more AG-UI events."""
from praisonaiagents.streaming.events import StreamEventType
if event.type == StreamEventType.DELTA_TEXT and event.content:
if not buffer.current_text_message_id:
buffer.start_text_message()
return [
TextMessageStartEvent(message_id=buffer.current_text_message_id, role="assistant"),
TextMessageContentEvent(
message_id=buffer.current_text_message_id, delta=event.content
),
]
return [
TextMessageContentEvent(
message_id=buffer.current_text_message_id, delta=event.content
)
]
if event.type == StreamEventType.TOOL_CALL_START and event.tool_call:
tool_call_id = event.tool_call.get("id") or str(uuid.uuid4())
tool_name = event.tool_call.get("name", "tool")
arguments = event.tool_call.get("arguments", {})
args_str = json.dumps(arguments) if isinstance(arguments, dict) else str(arguments)
parent_id = buffer.get_parent_message_id_for_tool_call() or message_id
buffer.start_tool_call(tool_call_id)
return list(create_tool_call_events(tool_call_id, tool_name, args_str, parent_id))
if event.type == StreamEventType.TOOL_CALL_RESULT and event.tool_call:
tool_call_id = event.tool_call.get("id") or str(uuid.uuid4())
raw_result = event.tool_call.get("result")
content = event.content if event.content is not None else str(raw_result or "")
buffer.end_tool_call(tool_call_id)
events: List[BaseEvent] = [
create_tool_result_event(tool_call_id, str(content), message_id)
]
if isinstance(raw_result, dict) and raw_result.get("mime_type") == A2UI_MIME_TYPE:
events.append(
CustomEvent(
name="a2ui",
value={
"mime_type": A2UI_MIME_TYPE,
"messages": raw_result.get("messages", []),
"surface_id": _infer_a2ui_surface_id(raw_result),
},
)
)
return events
return []
def _infer_a2ui_surface_id(result: Dict[str, Any]) -> str:
"""Infer surface id from send_a2ui_messages result (inline, no UI-layer deps)."""
for msg in result.get("messages") or []:
if isinstance(msg, dict):
create = msg.get("createSurface")
if isinstance(create, dict):
sid = create.get("surfaceId") or create.get("id")
if sid:
return str(sid)
sid = result.get("surface_id") or result.get("surfaceId")
return str(sid) if sid else "main"
async def async_stream_agent_response(
agent,
user_input: str,
@@ -351,63 +278,37 @@ async def async_stream_agent_response(
Yields:
AG-UI events
"""
import asyncio
# Emit run started
yield create_run_started_event(thread_id, run_id)
loop = asyncio.get_running_loop()
queue: asyncio.Queue = asyncio.Queue()
buffer = EventBuffer()
message_id = str(uuid.uuid4())
text_started = False
text_ended = False
original_emitter = getattr(agent, "stream_emitter", None)
def on_stream_event(event: Any) -> None:
loop.call_soon_threadsafe(queue.put_nowait, ("stream", event))
if original_emitter is not None:
original_emitter.add_callback(on_stream_event)
async def run_chat() -> Any:
try:
if hasattr(agent, "achat"):
return await agent.achat(user_input, stream=True)
if hasattr(agent, "chat"):
return await loop.run_in_executor(
None, lambda: agent.chat(user_input, stream=True)
)
raise ValueError("Agent must have 'chat' or 'achat' method")
finally:
loop.call_soon_threadsafe(queue.put_nowait, ("done", None))
chat_task = asyncio.create_task(run_chat())
try:
while True:
kind, payload = await queue.get()
if kind == "done":
break
for agui_event in stream_event_to_agui_events(payload, message_id, buffer):
if isinstance(agui_event, TextMessageStartEvent):
text_started = True
yield agui_event
response = await chat_task
if text_started and buffer.current_text_message_id and not text_ended:
yield TextMessageEndEvent(message_id=buffer.current_text_message_id)
text_ended = True
elif not text_started and response:
yield TextMessageStartEvent(message_id=message_id, role="assistant")
# Check if agent has async chat method
if hasattr(agent, 'achat'):
response = await agent.achat(user_input)
elif hasattr(agent, 'chat'):
# Run sync chat in executor
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(None, agent.chat, user_input)
else:
raise ValueError("Agent must have 'chat' or 'achat' method")
# Emit text message events
yield TextMessageStartEvent(message_id=message_id, role="assistant")
if response:
yield TextMessageContentEvent(message_id=message_id, delta=str(response))
yield TextMessageEndEvent(message_id=message_id)
yield TextMessageEndEvent(message_id=message_id)
# Emit run finished
yield create_run_finished_event(thread_id, run_id)
except Exception as e:
yield create_run_error_event(str(e))
finally:
if original_emitter is not None:
original_emitter.remove_callback(on_stream_event)
async def async_stream_agents_response(
@@ -430,6 +331,8 @@ async def async_stream_agents_response(
Yields:
AG-UI events
"""
import asyncio
# Emit run started
yield create_run_started_event(thread_id, run_id)
@@ -441,7 +344,7 @@ async def async_stream_agents_response(
result = await agents.astart(user_input)
elif hasattr(agents, 'start'):
# Run sync start in executor
loop = asyncio.get_running_loop()
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, agents.start, user_input)
else:
raise ValueError("Agents must have 'start' or 'astart' method")
@@ -133,13 +133,6 @@ class StateSnapshotEvent(BaseEvent):
snapshot: Dict[str, Any]
class CustomEvent(BaseEvent):
"""Custom extension event (e.g. structured A2UI payloads)."""
type: Literal["CUSTOM"] = "CUSTOM"
name: str
value: Any
# Message Types
class FunctionCall(BaseModel):
"""Function call in a tool call."""
@@ -182,7 +175,6 @@ Event = Union[
ToolCallEndEvent,
ToolCallResultEvent,
StateSnapshotEvent,
CustomEvent,
RunStartedEvent,
RunFinishedEvent,
RunErrorEvent,
@@ -1,26 +0,0 @@
"""
UI protocol types — lightweight contracts for agent-to-UI integration.
Zero runtime cost: TypedDict and constants only. No optional dependencies.
"""
from __future__ import annotations
from typing import Any, Dict, List, TypedDict, Union
# Standard MIME type for A2UI payloads (Google A2UI / A2A transport).
A2UI_MIME_TYPE = "application/json+a2ui"
class A2UIToolResultProtocol(TypedDict, total=False):
"""
Output shape from ``send_a2ui_messages`` — the integrator contract for any UI.
Example:
if result.get("mime_type") == A2UI_MIME_TYPE:
render(result["messages"])
"""
mime_type: str
messages: List[Dict[str, Any]]
a2ui_part: Union[Dict[str, Any], List[Any]]
+2 -20
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.6.48"
version = "1.6.37"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -101,22 +101,6 @@ crawl = [
"playwright>=1.47.0",
]
# Sandbox dependencies
sandbox = [
"e2b-code-interpreter>=1.0.0",
]
# Declarative agent UI (Google A2UI SDK — lazy loaded)
a2ui = [
"a2ui-agent-sdk>=0.2.1",
"a2a-sdk>=0.3.0,<1.0.0",
]
# Docker sandbox dependencies
sandbox-docker = [
"docker>=7.0.0",
]
# Combined features
all = [
"praisonaiagents[memory]",
@@ -128,12 +112,10 @@ all = [
"praisonaiagents[telemetry]",
"praisonaiagents[mongodb]",
"praisonaiagents[auth]",
"praisonaiagents[sandbox]",
"praisonaiagents[search]",
"praisonaiagents[crawl]",
"praisonaiagents[autonomy]",
"praisonaiagents[os]",
"praisonaiagents[a2ui]",
"praisonaiagents[os]"
]
[tool.setuptools.packages.find]
@@ -233,47 +233,6 @@ class TestA2AJsonRpc:
assert result["status"]["state"] == "completed"
assert result["artifacts"] is not None
assert len(result["artifacts"]) > 0
def test_post_a2a_message_send_a2ui_artifact(self):
"""message/send with A2UI tool result in chat_history → DataPart artifact."""
import json
from fastapi.testclient import TestClient
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
app, a2a = _make_app()
a2ui_payload = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
"a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
}
a2a.agent.chat = MagicMock(return_value="Surface ready.")
a2a.agent.chat_history = [
{"role": "tool", "content": json.dumps(a2ui_payload)},
]
client = TestClient(app)
response = client.post("/a2a", json={
"jsonrpc": "2.0",
"method": "message/send",
"id": "a2ui-1",
"params": {
"message": {
"messageId": "msg-a2ui",
"role": "user",
"parts": [{"text": "Show UI"}],
}
},
})
assert response.status_code == 200
result = response.json()["result"]
artifact = result["artifacts"][0]
parts = artifact["parts"]
data_parts = [p for p in parts if "data" in p]
assert data_parts, "Expected DataPart in artifact"
assert data_parts[0]["metadata"]["mimeType"] == A2UI_MIME_TYPE
def test_post_a2a_message_send_creates_task_in_store(self):
"""message/send should create a task in TaskStore."""
@@ -1,66 +0,0 @@
"""Tests for A2UI artifact conversion in A2A transport."""
import json
import pytest
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
SAMPLE_A2UI_RESULT = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
"a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
}
class TestFindA2uiToolResult:
def test_finds_last_a2ui_tool_message(self):
from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
history = [
{"role": "user", "content": "show ui"},
{"role": "tool", "content": json.dumps({"mime_type": "text/plain", "data": "x"})},
{"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)},
]
result = find_a2ui_tool_result(history)
assert result is not None
assert result["mime_type"] == A2UI_MIME_TYPE
def test_returns_none_without_a2ui(self):
from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
assert find_a2ui_tool_result([]) is None
assert find_a2ui_tool_result([{"role": "tool", "content": "plain text"}]) is None
class TestCreateA2uiArtifact:
def test_creates_datapart_with_mime(self):
from praisonaiagents.ui.a2a.conversion import create_a2ui_artifact
from praisonaiagents.ui.a2a.types import DataPart
artifact = create_a2ui_artifact(SAMPLE_A2UI_RESULT, name="ui-surface")
assert artifact.name == "ui-surface"
assert len(artifact.parts) == 1
part = artifact.parts[0]
assert isinstance(part, DataPart)
assert part.metadata.get("mimeType") == A2UI_MIME_TYPE
assert "messages" in part.data
class TestCreateResponseArtifact:
def test_prefers_a2ui_from_history(self):
from praisonaiagents.ui.a2a.conversion import create_response_artifact
from praisonaiagents.ui.a2a.types import DataPart
history = [{"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)}]
artifact = create_response_artifact("Done.", chat_history=history)
assert isinstance(artifact.parts[0], DataPart)
def test_falls_back_to_text(self):
from praisonaiagents.ui.a2a.conversion import create_response_artifact
from praisonaiagents.ui.a2a.types import TextPart
artifact = create_response_artifact("Hello", chat_history=[])
assert isinstance(artifact.parts[0], TextPart)
assert artifact.parts[0].text == "Hello"
@@ -0,0 +1 @@
"""A2UI Tests Package."""
@@ -1,51 +0,0 @@
"""Tests for send_a2ui_messages minimal arg unwrap."""
import json
import pytest
from praisonaiagents.tools.a2ui_tools import _unwrap_messages
try:
import a2ui # noqa: F401
HAS_A2UI_SDK = True
except ImportError:
HAS_A2UI_SDK = False
SAMPLE_MSG = {"createSurface": {"surfaceId": "main", "catalogId": "basic"}}
class TestUnwrapMessages:
def test_list_input(self):
assert _unwrap_messages([SAMPLE_MSG]) == [SAMPLE_MSG]
def test_json_string_list(self):
assert _unwrap_messages(json.dumps([SAMPLE_MSG])) == [SAMPLE_MSG]
def test_wrapped_messages_dict(self):
assert _unwrap_messages({"messages": [SAMPLE_MSG]}) == [SAMPLE_MSG]
def test_json_string_wrapped_object(self):
assert _unwrap_messages(json.dumps({"messages": [SAMPLE_MSG]})) == [SAMPLE_MSG]
def test_single_message_dict(self):
assert _unwrap_messages(SAMPLE_MSG) == [SAMPLE_MSG]
def test_invalid_json_raises(self):
with pytest.raises(ValueError, match="not valid JSON"):
_unwrap_messages("{not json")
def test_invalid_shape_raises(self):
with pytest.raises(ValueError, match="must be"):
_unwrap_messages(123)
@pytest.mark.skipif(not HAS_A2UI_SDK, reason="a2ui-agent-sdk not installed")
class TestSendA2uiMessagesIntegration:
def test_full_tool_call(self):
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
result = send_a2ui_messages(messages=[SAMPLE_MSG])
assert result["mime_type"] == "application/json+a2ui"
assert len(result["messages"]) == 1
@@ -1,60 +0,0 @@
"""Tests for A2UI adapter (optional a2ui-agent-sdk dependency)."""
import pytest
class TestA2UIAdapterImport:
def test_a2ui_class_export(self):
from praisonaiagents.ui.a2ui import A2UI
assert A2UI is not None
assert hasattr(A2UI, "create_part")
def test_ui_lazy_a2ui_export(self):
from praisonaiagents.ui import A2UI
assert A2UI is not None
def test_missing_optional_dep_raises_clear_error(self):
pytest.importorskip("a2ui", reason="a2ui-agent-sdk not installed")
# If installed, adapter imports should work
from praisonaiagents.ui.a2ui.adapter import create_a2ui_part
part = create_a2ui_part({"createSurface": {"surfaceId": "main", "catalogId": "test"}})
assert part is not None
class TestA2UITool:
def test_send_a2ui_messages_requires_sdk(self):
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
if pytest.importorskip("a2ui", reason="skip without sdk"):
result = send_a2ui_messages(
messages=[{"createSurface": {"surfaceId": "x", "catalogId": "c"}}]
)
assert result["mime_type"] == "application/json+a2ui"
assert len(result["messages"]) == 1
class TestA2UIAdapterSmoke:
def test_is_a2ui_part_with_sdk(self):
pytest.importorskip("a2ui", reason="a2ui-agent-sdk not installed")
from praisonaiagents.ui.a2ui.adapter import create_a2ui_part, is_a2ui_part
part = create_a2ui_part({"messages": [{"createSurface": {"surfaceId": "m"}}]})
assert is_a2ui_part(part) is True
def test_missing_sdk_import_error_message(self):
from unittest.mock import patch
from praisonaiagents.ui.a2ui import adapter
def _raise(_name):
raise ImportError(
"A2UI support requires the optional dependency. "
"Install with: pip install praisonaiagents[a2ui]"
)
with patch.object(adapter, "_import", side_effect=_raise):
with pytest.raises(ImportError, match="praisonaiagents\\[a2ui\\]"):
adapter.create_a2ui_part({"messages": []})
@@ -0,0 +1,202 @@
"""
Tests for A2UI Agent Integration
TDD: Write tests first, then implement the A2UIAgent class.
"""
import os
import pytest
# Skip tests if no API key
pytestmark = pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set"
)
class TestA2UIAgent:
"""Test A2UIAgent wrapper class."""
def test_a2ui_agent_creation(self):
"""Test A2UIAgent creation with an agent."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(
name="TestAgent",
role="Test assistant",
goal="Help with testing"
)
a2ui_agent = A2UIAgent(agent=agent)
assert a2ui_agent.agent is agent
assert a2ui_agent.surface_id is not None
def test_a2ui_agent_with_custom_surface_id(self):
"""Test A2UIAgent with custom surface ID."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent, surface_id="custom-surface")
assert a2ui_agent.surface_id == "custom-surface"
def test_render_text_response(self):
"""Test rendering a simple text response as A2UI."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
# Render a simple text response
messages = a2ui_agent.render_text("Hello, World!")
assert isinstance(messages, list)
assert len(messages) >= 2 # createSurface + updateComponents
# Check createSurface
assert "createSurface" in messages[0]
# Check updateComponents has a text component
update_msg = next((m for m in messages if "updateComponents" in m), None)
assert update_msg is not None
components = update_msg["updateComponents"]["components"]
assert any(c.get("component") == "Text" for c in components)
def test_render_list_response(self):
"""Test rendering a list of items as A2UI cards."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
items = [
{"title": "Item 1", "description": "First item"},
{"title": "Item 2", "description": "Second item"},
]
messages = a2ui_agent.render_list(
title="My Items",
items=items,
item_title_key="title",
item_description_key="description"
)
assert isinstance(messages, list)
assert len(messages) >= 2
def test_render_form(self):
"""Test rendering a form as A2UI."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
fields = [
{"id": "name", "label": "Name", "type": "text"},
{"id": "email", "label": "Email", "type": "text"},
]
messages = a2ui_agent.render_form(
title="Contact Form",
fields=fields,
submit_action="submit_form"
)
assert isinstance(messages, list)
assert len(messages) >= 2
# Check for TextField components
update_msg = next((m for m in messages if "updateComponents" in m), None)
assert update_msg is not None
components = update_msg["updateComponents"]["components"]
text_fields = [c for c in components if c.get("component") == "TextField"]
assert len(text_fields) >= 2
def test_get_surface(self):
"""Test getting the underlying Surface object."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
from praisonaiagents.ui.a2ui.surface import Surface
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
surface = a2ui_agent.get_surface()
assert isinstance(surface, Surface)
assert surface.surface_id == a2ui_agent.surface_id
class TestA2UIAgentWithLLM:
"""Test A2UIAgent with actual LLM calls."""
@pytest.mark.integration
@pytest.mark.asyncio
async def test_chat_with_ui(self):
"""Test chat that returns A2UI response."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
agent = Agent(
name="UIAgent",
role="UI Generator",
goal="Generate helpful UI responses",
llm="gpt-4o-mini"
)
a2ui_agent = A2UIAgent(agent=agent)
# This should work with the agent's chat
response = await a2ui_agent.chat_async("Say hello")
assert response is not None
# Response should contain text
assert isinstance(response, str) or isinstance(response, dict)
class TestA2UIAgentIntegration:
"""Integration tests for A2UIAgent with A2A."""
def test_get_a2a_data_part(self):
"""Test getting A2UI as A2A DataPart."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
from praisonaiagents.ui.a2ui.extension import is_a2ui_part
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
# Render something
a2ui_agent.render_text("Hello")
# Get as A2A DataPart
part = a2ui_agent.to_a2a_part()
assert part is not None
assert is_a2ui_part(part)
def test_to_json(self):
"""Test getting A2UI as JSON string."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui.agent import A2UIAgent
import json
agent = Agent(name="TestAgent", role="Test", goal="Test")
a2ui_agent = A2UIAgent(agent=agent)
# Render something
a2ui_agent.render_text("Hello")
# Get as JSON
json_str = a2ui_agent.to_json()
# Should be valid JSON
parsed = json.loads(json_str)
assert isinstance(parsed, list)
@@ -0,0 +1,115 @@
"""
Tests for A2UI Extension Module
TDD: Write tests first, then implement the extension.
"""
class TestA2UIExtension:
"""Test A2UI extension helpers for A2A integration."""
def test_a2ui_mime_type(self):
"""Test A2UI MIME type constant."""
from praisonaiagents.ui.a2ui.extension import A2UI_MIME_TYPE
assert A2UI_MIME_TYPE == "application/json+a2ui"
def test_a2ui_extension_uri(self):
"""Test A2UI extension URI constant."""
from praisonaiagents.ui.a2ui.extension import A2UI_EXTENSION_URI
assert "a2ui" in A2UI_EXTENSION_URI.lower()
def test_create_a2ui_part(self):
"""Test create_a2ui_part wraps A2UI data in A2A DataPart."""
from praisonaiagents.ui.a2ui.extension import create_a2ui_part, A2UI_MIME_TYPE
a2ui_data = {
"createSurface": {
"surfaceId": "main",
"catalogId": "https://a2ui.dev/standard"
}
}
part = create_a2ui_part(a2ui_data)
# Should return a Part with DataPart root
assert part is not None
assert hasattr(part, 'data')
assert part.data == a2ui_data
assert part.metadata is not None
assert part.metadata.get("mimeType") == A2UI_MIME_TYPE
def test_is_a2ui_part_true(self):
"""Test is_a2ui_part returns True for A2UI parts."""
from praisonaiagents.ui.a2ui.extension import create_a2ui_part, is_a2ui_part
a2ui_data = {"createSurface": {"surfaceId": "main", "catalogId": "test"}}
part = create_a2ui_part(a2ui_data)
assert is_a2ui_part(part) is True
def test_is_a2ui_part_false(self):
"""Test is_a2ui_part returns False for non-A2UI parts."""
from praisonaiagents.ui.a2ui.extension import is_a2ui_part
from praisonaiagents.ui.a2ui.types import A2UIDataPart
# Create a regular data part without A2UI mime type
part = A2UIDataPart(data={"foo": "bar"}, metadata={"mimeType": "application/json"})
assert is_a2ui_part(part) is False
def test_get_a2ui_agent_extension(self):
"""Test get_a2ui_agent_extension returns AgentExtension config."""
from praisonaiagents.ui.a2ui.extension import get_a2ui_agent_extension
ext = get_a2ui_agent_extension()
assert ext is not None
assert hasattr(ext, 'uri')
assert "a2ui" in ext.uri.lower()
assert hasattr(ext, 'description')
def test_get_a2ui_agent_extension_with_inline_catalog(self):
"""Test get_a2ui_agent_extension with inline catalog support."""
from praisonaiagents.ui.a2ui.extension import get_a2ui_agent_extension
ext = get_a2ui_agent_extension(accepts_inline_custom_catalog=True)
assert ext.params is not None
assert ext.params.get("acceptsInlineCustomCatalog") is True
def test_standard_catalog_id(self):
"""Test standard catalog ID constant."""
from praisonaiagents.ui.a2ui.extension import STANDARD_CATALOG_ID
assert "standard_catalog" in STANDARD_CATALOG_ID.lower() or "a2ui" in STANDARD_CATALOG_ID.lower()
class TestA2UIDataPart:
"""Test A2UI DataPart type."""
def test_a2ui_data_part_creation(self):
"""Test A2UIDataPart creation."""
from praisonaiagents.ui.a2ui.types import A2UIDataPart
part = A2UIDataPart(
data={"test": "value"},
metadata={"mimeType": "application/json+a2ui"}
)
assert part.data == {"test": "value"}
assert part.metadata["mimeType"] == "application/json+a2ui"
def test_a2ui_data_part_to_dict(self):
"""Test A2UIDataPart serialization."""
from praisonaiagents.ui.a2ui.types import A2UIDataPart
part = A2UIDataPart(
data={"createSurface": {"surfaceId": "main"}},
metadata={"mimeType": "application/json+a2ui"}
)
data = part.to_dict()
assert "data" in data
assert "metadata" in data
@@ -0,0 +1,225 @@
"""
End-to-End Integration Tests for A2UI
Tests the complete A2UI workflow from agent to JSON output.
"""
import json
import os
import pytest
class TestA2UIEndToEnd:
"""End-to-end integration tests."""
def test_complete_surface_workflow(self):
"""Test complete workflow: create surface, add components, generate JSON."""
from praisonaiagents.ui.a2ui import (
Surface,
PathBinding,
)
# Create surface
surface = Surface(surface_id="test-surface")
# Add components using fluent API
surface.text(id="title", text="Welcome", usage_hint="h1")
surface.text(id="subtitle", text=PathBinding(path="/subtitle"), usage_hint="h2")
surface.button(
id="action-btn",
child="btn-text",
action_name="do_action",
action_context=[{"key": "id", "value": "123"}],
primary=True
)
surface.text(id="btn-text", text="Click Me")
surface.column(id="root", children=["title", "subtitle", "action-btn"])
# Set data
surface.set_data("subtitle", "Hello World")
# Generate messages
messages = surface.to_messages()
# Verify structure
assert len(messages) == 3 # createSurface, updateComponents, updateDataModel
# Verify createSurface
assert "createSurface" in messages[0]
assert messages[0]["createSurface"]["surfaceId"] == "test-surface"
# Verify updateComponents
assert "updateComponents" in messages[1]
components = messages[1]["updateComponents"]["components"]
assert len(components) == 5 # title, subtitle, btn, btn-text, root
# Verify updateDataModel
assert "updateDataModel" in messages[2]
assert messages[2]["updateDataModel"]["value"]["subtitle"] == "Hello World"
# Verify JSON output
json_str = surface.to_json()
parsed = json.loads(json_str)
assert len(parsed) == 3
def test_a2ui_agent_render_workflow(self):
"""Test A2UIAgent rendering workflow."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui import A2UIAgent
agent = Agent(
name="TestAgent",
role="Test assistant",
goal="Help with testing"
)
a2ui_agent = A2UIAgent(agent=agent, surface_id="agent-surface")
# Render text
messages = a2ui_agent.render_text(
text="This is a test response",
title="Test Title"
)
# Verify structure
assert len(messages) >= 2
assert "createSurface" in messages[0]
# Verify JSON export
json_str = a2ui_agent.to_json()
parsed = json.loads(json_str)
assert isinstance(parsed, list)
def test_template_workflow(self):
"""Test template-based workflow."""
from praisonaiagents.ui.a2ui import (
ChatTemplate,
ListTemplate,
FormTemplate,
DashboardTemplate,
)
# Chat template
chat = ChatTemplate(surface_id="chat")
chat.add_user_message("Hello!")
chat.add_agent_message("Hi there!")
chat_messages = chat.to_messages()
assert len(chat_messages) >= 2
# List template
list_template = ListTemplate(surface_id="list", title="Items")
list_template.add_item(title="Item 1", description="First")
list_template.add_item(title="Item 2", description="Second")
list_messages = list_template.to_messages()
assert len(list_messages) >= 2
# Form template
form = FormTemplate(surface_id="form", title="Contact")
form.add_text_field(id="name", label="Name")
form.add_email_field(id="email", label="Email")
form.set_submit_action("submit", "Send")
form_messages = form.to_messages()
assert len(form_messages) >= 2
# Dashboard template
dashboard = DashboardTemplate(surface_id="dashboard", title="Dashboard")
dashboard.add_panel(id="stats", title="Stats", content="100 items")
dashboard.add_panel(id="chart", title="Chart", content="[Chart]")
dashboard_messages = dashboard.to_messages()
assert len(dashboard_messages) >= 2
def test_a2a_integration(self):
"""Test A2A protocol integration."""
from praisonaiagents.ui.a2ui import (
Surface,
create_a2ui_part,
is_a2ui_part,
A2UI_MIME_TYPE,
)
# Create surface
surface = Surface(surface_id="a2a-test")
surface.text(id="msg", text="Hello from A2A")
surface.column(id="root", children=["msg"])
# Get messages
messages = surface.to_messages()
# Wrap in A2A DataPart
part = create_a2ui_part({"messages": messages})
# Verify A2UI part
assert is_a2ui_part(part)
assert part.metadata["mimeType"] == A2UI_MIME_TYPE
assert "messages" in part.data
assert len(part.data["messages"]) >= 2
def test_data_binding_workflow(self):
"""Test data binding with path references."""
from praisonaiagents.ui.a2ui import (
Surface,
PathBinding,
ChildrenTemplate,
)
surface = Surface(surface_id="data-binding-test")
# Add components with data binding
surface.text(id="title", text=PathBinding(path="/title"), usage_hint="h1")
surface.text(id="item-template", text=PathBinding(path="/name"))
# Add list with template
surface.list(
id="items",
children=ChildrenTemplate(component_id="item-template", path="/items"),
direction="vertical"
)
surface.column(id="root", children=["title", "items"])
# Set data
surface.set_data("title", "My List")
surface.set_data("items", [
{"name": "Item 1"},
{"name": "Item 2"},
{"name": "Item 3"},
])
messages = surface.to_messages()
# Verify data model
data_msg = next((m for m in messages if "updateDataModel" in m), None)
assert data_msg is not None
assert data_msg["updateDataModel"]["value"]["title"] == "My List"
assert len(data_msg["updateDataModel"]["value"]["items"]) == 3
@pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set"
)
class TestA2UIWithLLM:
"""Tests that require LLM API calls."""
def test_agent_chat_and_render(self):
"""Test agent chat followed by UI rendering."""
from praisonaiagents import Agent
from praisonaiagents.ui.a2ui import A2UIAgent
agent = Agent(
name="UIAgent",
role="UI Generator",
goal="Generate helpful responses",
llm="gpt-4o-mini"
)
a2ui_agent = A2UIAgent(agent=agent)
# Chat with agent
response = a2ui_agent.chat("What is 2+2?")
# Render response as UI
messages = a2ui_agent.render_text(response, title="Answer")
assert len(messages) >= 2
assert "createSurface" in messages[0]
@@ -1,27 +0,0 @@
"""Tests for UI protocol types (zero-dep contracts)."""
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
class TestA2UIProtocols:
def test_mime_type_constant(self):
assert A2UI_MIME_TYPE == "application/json+a2ui"
def test_tool_result_protocol_shape(self):
result: A2UIToolResultProtocol = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main"}}],
"a2ui_part": {"messages": []},
}
assert result["mime_type"] == A2UI_MIME_TYPE
def test_exported_from_a2ui_package(self):
from praisonaiagents.ui.a2ui import A2UI_MIME_TYPE, A2UIToolResultProtocol
assert A2UI_MIME_TYPE == "application/json+a2ui"
assert A2UIToolResultProtocol is not None
def test_top_level_lazy_a2ui_export(self):
from praisonaiagents import A2UI
assert A2UI is not None
Loaded 100 of 321 files, more files were not shown because too many files have changed in this diff. Show more