Compare commits

...
25 Commits
Author SHA1 Message Date
MervinPraison c427e2588e Release v0.2.32 2026-07-09 17:05:21 +00:00
praisonai-triage-agent[bot] 30719720cd Merge pull request #30 from MervinPraison/claude/issue-1593-20260501
feat: Add SwarmScore integration tool for agent trust ratings
2026-07-09 16:41:05 +00:00
Mervin Praison d4ac1e42ce Merge pull request #51 from MervinPraison/fix/merge-gate-stash-config
fix(merge-gate): stash gate-config.js with merge-gate.js
2026-07-09 16:21:23 +01:00
MervinPraisonandMervin Praison d8ee45d8b7 style(swarmscore): remove extraneous f-string prefixes in example (ruff F541)
Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 14:57:17 +00:00
MervinPraison 3bf058e687 fix(merge-gate): stash gate-config.js with merge-gate.js
After PR checkout, require(/tmp/merge-gate.js) failed looking for
./gate-config.js. Keep both scripts in /tmp/merge-gate-scripts/.
2026-07-09 15:57:13 +01:00
Mervin Praison 04f1d165ff Merge pull request #42 from MervinPraison/claude/issue-1887-20260610
feat: add Inkog security analysis tool for AI agent code
2026-07-09 15:56:21 +01:00
MervinPraisonandMervin Praison d8d3ef8df0 fix(swarmscore): default run() action to 'load' for graceful degradation
Every other multi-action tool in the repo provides a default for `action`.
Without it, a bare `tool.run()` raised a raw TypeError bypassing the
ToolResult(success=False) error path. Add default + regression test.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 13:42:35 +00:00
Mervin Praison 11f67da66f Merge pull request #50 from MervinPraison/fix/merge-gate-claude-action
fix(merge-gate): add claude-code-action and restore after PR checkout
2026-07-09 14:37:41 +01:00
MervinPraisonandMervin Praison 222787293c fix(swarmscore): URL-encode slug in load-by-slug path
Prevents malformed request paths when a slug contains special
characters (e.g. '/', spaces). Adds a regression test asserting
the encoded path.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 13:33:46 +00:00
MervinPraisonandMervin Praison 5353ea28d6 fix: guard analyze_findings against non-dict scan output
Move the "error" key guard inside the try block and short-circuit
non-dict (text/HTML) scan results to prevent an unhandled TypeError
when a string containing "error" is passed. Add regression test.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 13:32:24 +00:00
MervinPraisonandMervin Praison d873ccd07b fix: standardize Inkog CLI flags to single-dash across all scan methods
skill_scan and mcp_scan used double-dash flags (--deep, --output, --repo)
while scan_directory and the version check (inkog -version) use single-dash
Go-style flags. Align subcommands and docs on single-dash for consistency.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 13:32:24 +00:00
MervinPraisonandpraisonai-triage-agent[bot] 11ca0e8bde fix: honor InkogTool api_key in availability check and add severity to scan_agent_code
- _check_inkog_available now accepts an api_key arg so the constructor's
  api_key is respected instead of only reading INKOG_API_KEY env var
- scan_agent_code now forwards a severity parameter (matches docs/CI examples)
- add regression tests for both fixes

Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>
2026-07-09 13:32:24 +00:00
MervinPraison f39b01afda Fix ruff F541 in Inkog tool. 2026-07-09 13:32:24 +00:00
MervinPraison ef02203d46 Fix ruff lint in Inkog tool tests. 2026-07-09 13:32:24 +00:00
MervinPraison 5a08365e15 feat: add Inkog security analysis tool for AI agent code
- Add InkogTool class with comprehensive security analysis capabilities
- Support for token bombing, prompt injection, recursive tool calling detection
- Multi-format output: JSON, text, HTML, SARIF
- Compliance mapping to EU AI Act, NIST AI RMF, OWASP LLM Top 10
- Support for 21+ frameworks (LangChain, CrewAI, AutoGen, etc.)
- Deep scan with AI orchestrator analysis
- MCP server security auditing
- Skill package vulnerability scanning
- Comprehensive test suite and documentation
- CI/CD integration examples

Fixes MervinPraison/PraisonAI#1887
2026-07-09 13:32:24 +00:00
MervinPraison f21fca919d Release v0.2.31 2026-07-09 13:31:47 +00:00
Mervin Praison 70d8bc9bf6 Merge pull request #40 from MervinPraison/claude/issue-1716-20260522
feat: Add NEXUS Prediction Market Tool Integration
2026-07-09 14:30:43 +01:00
MervinPraison da6ce8387c fix(merge-gate): add claude-code-action and restore after PR checkout
Assess was failing because Tools had no local action, and checkout of
the PR branch would drop it even when present. Stash/restore from main.
2026-07-09 14:30:10 +01:00
Mervin Praison 1e08305c80 fix(release): stop failing on gitignored uv.lock (#49)
git add uv.lock exits 1 under bash -e and aborts the release before
publish. Only commit pyproject.toml (uv.lock is intentionally ignored).
2026-07-09 13:52:13 +01:00
MervinPraisonandMervin Praison dd7fe0efa4 fix(swarmscore): derive discovery manifest URL from domain root via urlparse
Replace fragile split("/v1/") host derivation in get_discovery_manifest
with urlparse-based origin (scheme+netloc), so .well-known resolves to the
domain root (RFC 8615) regardless of version prefix (/v1/, /v2/, /api/, none).
Adds a parametrized test covering multiple base-URL shapes.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-09 09:05:32 +00:00
MervinPraisonandMervin Praison f404b2e903 fix(nexus): resolve Greptile P1 issues in exports and tests
- Remove get_market_data / check_arbitrage_opportunities from tool_map
  and __all__ in tools/__init__.py; they are instance methods, not
  module-level names, so importing them raised AttributeError and
  polluted the package namespace with generic identifiers.
- Fix two failure-path tests that raised a plain Exception (never caught
  by the tool). Now raise requests.exceptions.ConnectionError and a
  proper HTTPError with response.status_code=402 to exercise the real
  error-handling branches.

Co-authored-by: Mervin Praison <[email protected]>
2026-07-08 22:32:09 +00:00
MervinPraisonandMervin Praison 94a0167653 fix(swarmscore): use ToolResult.output in docs/example; export in __all__
- Replace all result.data references with result.output in docs and example
  (ToolResult exposes .output, not .data)
- Add SwarmScoreTool and standalone functions to tools __all__ for wildcard imports

Co-authored-by: Mervin Praison <[email protected]>
2026-07-08 22:32:06 +00:00
MervinPraisonandMervin Praison 65a53de573 fix: align SwarmScore tool with ToolResult API and resolve rebase
- Use ToolResult(output=..., success=..., metadata=...) instead of
  non-existent data/message kwargs
- Call super().__init__() for BaseTool registration
- Derive discovery manifest URL from api_base_url instead of hardcoding
- Remove unused urllib.parse.urljoin import
- Add swarmscore optional-dependency group (requests) to match docs
- Fix tests to mock real requests.exceptions namespace and read result.output

Co-authored-by: Mervin Praison <[email protected]>
2026-07-08 13:41:23 +00:00
MervinPraisonandpraisonai-triage-agent[bot] 8a47aff40a feat: add NEXUS prediction market tool integration (fixes MervinPraison/PraisonAI#1716)
- Added NexusPredictionMarketTool for live Kalshi/Polymarket data access
- Supports free market data queries and paid arbitrage analysis
- Includes comprehensive examples and documentation
- Full test coverage with mocked API responses
- Integrated with PraisonAI agent workflow patterns

Features:
- Real-time prediction market data from NEXUS API
- Arbitrage opportunity detection (paid feature: /bin/bash.02 USDC)
- Agent metadata and capabilities discovery
- Source tracking via X-NEXUS-Source header
- Error handling for network/payment issues

API Endpoints:
- Free: /kalshi?market={market_name}
- Paid: /arb/check?markets={market1,market2}
- Validation: https://agentic.market/validate

Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>
2026-07-08 14:39:21 +01:00
MervinPraisonandpraisonai-triage-agent[bot] 6c25336a1b feat: add SwarmScore integration tool for PraisonAI (fixes MervinPraison/PraisonAI#1593)
Add SwarmScore trust rating integration to PraisonAI-Tools:

- SwarmScoreTool class for loading and verifying agent trust scores
- Standalone utility functions for easy integration
- Comprehensive documentation and examples
- Complete test coverage
- Integration with PraisonAI agent workflows

SwarmScore provides portable, cryptographically signed trust ratings
based on verified execution history and success rates.

Co-authored-by: praisonai-triage-agent[bot] <praisonai-triage-agent[bot]@users.noreply.github.com>
2026-07-08 13:38:01 +00:00
19 changed files with 4225 additions and 7 deletions

No files matched your search

@@ -0,0 +1,290 @@
name: "Claude Code Action"
description: "Run Claude Code in GitHub Actions workflows"
inputs:
anthropic_api_key:
description: "Anthropic API key"
required: false
claude_code_oauth_token:
description: "Claude Code OAuth Token for Claude Pro"
required: false
github_token:
description: "GitHub token for Claude to operate with"
required: false
default: ${{ github.token }}
trigger_phrase:
description: "The trigger phrase to look for in comments, issue/PR bodies, and issue titles"
required: false
default: "@claude"
assignee_trigger:
description: "The assignee username that triggers the action (e.g. @claude). Only used for issue assignment"
required: false
max_turns:
description: "Maximum number of conversation turns Claude can take"
required: false
timeout_minutes:
description: "Timeout in minutes for execution"
required: false
default: "30"
model:
description: "Model to use (provider-specific format required for Bedrock/Vertex)"
required: false
use_bedrock:
description: "Use Amazon Bedrock with OIDC authentication instead of direct Anthropic API"
required: false
default: "false"
use_vertex:
description: "Use Google Vertex AI with OIDC authentication instead of direct Anthropic API"
required: false
default: "false"
allowed_tools:
description: "Additional tools for Claude to use (the base GitHub tools will always be included)"
required: false
default: ""
disallowed_tools:
description: "Tools that Claude should never use"
required: false
default: ""
custom_instructions:
description: "Additional custom instructions to include in the prompt for Claude"
required: false
default: ""
mcp_config:
description: "Additional MCP configuration (JSON string) that merges with the built-in GitHub MCP servers"
required: false
default: ""
claude_env:
description: "Custom environment variables to pass to Claude Code execution (YAML format)"
required: false
default: ""
direct_prompt:
description: "Direct prompt for Claude to execute automatically without needing a trigger (for automated workflows)"
required: false
runs:
using: "composite"
steps:
- name: Install Claude Code
shell: bash
run: npm install -g @anthropic-ai/claude-code
- name: Install GitHub MCP Server
shell: bash
run: |
claude mcp add-json github '{
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-e",
"GITHUB_PERSONAL_ACCESS_TOKEN",
"ghcr.io/github/github-mcp-server:sha-ff3036d"
],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "${{ inputs.github_token }}"
}
}'
- name: Extract GitHub Context and Create Prompt
shell: bash
id: prepare_context
env:
# Pass user-controlled inputs via environment variables to prevent script injection (GHSL-2025-093)
EVENT_COMMENT_BODY: ${{ github.event.comment.body }}
EVENT_ISSUE_TITLE: ${{ github.event.issue.title }}
EVENT_ISSUE_BODY: ${{ github.event.issue.body }}
EVENT_ISSUE_NUMBER: ${{ github.event.issue.number }}
EVENT_PR_TITLE: ${{ github.event.pull_request.title }}
EVENT_PR_NUMBER: ${{ github.event.pull_request.number }}
EVENT_REVIEW_BODY: ${{ github.event.review.body }}
EVENT_ASSIGNEE_LOGIN: ${{ github.event.assignee.login }}
INPUT_TRIGGER_PHRASE: ${{ inputs.trigger_phrase }}
INPUT_ASSIGNEE_TRIGGER: ${{ inputs.assignee_trigger }}
INPUT_DIRECT_PROMPT: ${{ inputs.direct_prompt }}
EVENT_ACTION: ${{ github.event.action }}
EVENT_NAME: ${{ github.event_name }}
REPO_NAME: ${{ github.repository }}
run: |
echo "🔍 Extracting GitHub context from event: $EVENT_NAME"
# Function to check for trigger phrase
check_trigger() {
local text="$1"
local trigger="$INPUT_TRIGGER_PHRASE"
if [[ "$text" == *"$trigger"* ]]; then
return 0
fi
return 1
}
# Extract context based on event type
TRIGGER_FOUND="false"
USER_REQUEST=""
CONTEXT_INFO=""
case "$EVENT_NAME" in
"issue_comment")
COMMENT_BODY="$EVENT_COMMENT_BODY"
ISSUE_TITLE="$EVENT_ISSUE_TITLE"
ISSUE_NUMBER="$EVENT_ISSUE_NUMBER"
if check_trigger "$COMMENT_BODY"; then
TRIGGER_FOUND="true"
USER_REQUEST="$COMMENT_BODY"
CONTEXT_INFO="Issue Comment on #$ISSUE_NUMBER: $ISSUE_TITLE"
fi
;;
"pull_request_review_comment")
COMMENT_BODY="$EVENT_COMMENT_BODY"
PR_TITLE="$EVENT_PR_TITLE"
PR_NUMBER="$EVENT_PR_NUMBER"
if check_trigger "$COMMENT_BODY"; then
TRIGGER_FOUND="true"
USER_REQUEST="$COMMENT_BODY"
CONTEXT_INFO="PR Comment on #$PR_NUMBER: $PR_TITLE"
fi
;;
"pull_request_review")
REVIEW_BODY="$EVENT_REVIEW_BODY"
PR_TITLE="$EVENT_PR_TITLE"
PR_NUMBER="$EVENT_PR_NUMBER"
if check_trigger "$REVIEW_BODY"; then
TRIGGER_FOUND="true"
USER_REQUEST="$REVIEW_BODY"
CONTEXT_INFO="PR Review on #$PR_NUMBER: $PR_TITLE"
fi
;;
"issues")
ISSUE_BODY="$EVENT_ISSUE_BODY"
ISSUE_TITLE="$EVENT_ISSUE_TITLE"
ISSUE_NUMBER="$EVENT_ISSUE_NUMBER"
if check_trigger "$ISSUE_TITLE" || check_trigger "$ISSUE_BODY"; then
TRIGGER_FOUND="true"
USER_REQUEST="$ISSUE_BODY"
CONTEXT_INFO="Issue #$ISSUE_NUMBER: $ISSUE_TITLE"
elif [[ "$EVENT_ACTION" == "assigned" && -n "$INPUT_ASSIGNEE_TRIGGER" ]]; then
ASSIGNEE="$EVENT_ASSIGNEE_LOGIN"
if [[ "$ASSIGNEE" == "$INPUT_ASSIGNEE_TRIGGER" ]]; then
TRIGGER_FOUND="true"
USER_REQUEST="$ISSUE_BODY"
CONTEXT_INFO="Issue #$ISSUE_NUMBER assigned to $ASSIGNEE: $ISSUE_TITLE"
fi
fi
;;
esac
# Check for direct prompt override
if [[ -n "$INPUT_DIRECT_PROMPT" ]]; then
TRIGGER_FOUND="true"
USER_REQUEST="$INPUT_DIRECT_PROMPT"
CONTEXT_INFO="Automated GitHub workflow"
fi
if [[ "$TRIGGER_FOUND" != "true" ]]; then
echo "❌ No trigger phrase found or direct prompt provided. Exiting gracefully."
echo "SKIP_EXECUTION=true" >> $GITHUB_ENV
exit 0
fi
echo "✅ Trigger found! Context: $CONTEXT_INFO"
# Create comprehensive prompt
mkdir -p /tmp/claude-action
cat > /tmp/claude-action/github-context-prompt.txt << EOF
You are Claude Code, an AI assistant helping with GitHub workflows and code.
Repository: $REPO_NAME
Context: $CONTEXT_INFO
Event: $EVENT_NAME
User Request:
$USER_REQUEST
Please analyze the request and provide helpful assistance.
CRITICAL INSTRUCTION: If your task involves modifying codebase files to add a feature or fix a bug:
STEP 0 — SETUP GIT IDENTITY:
git config user.name "MervinPraison"
git config user.email "[email protected]"
STEP 1 — READ GUIDELINES:
MANDATORY: Before writing any code, you MUST carefully read "src/praisonai-agents/AGENTS.md" to understand this repository's extremely specific Architecture, Protocol-First design rules, and Agent-Centric Philosophy.
STEP 2 — ARCHITECTURE VALIDATION & ROUTING (MANDATORY before writing code):
Before implementing anything, answer these questions:
- CORE vs WRAPPER vs TOOLS ROUTING:
1. Core SDK (praisonaiagents/): Only core protocols, base classes, decorators. No heavy implementations.
2. Wrapper (praisonai/): CLI, heavy implementations, optional dependencies.
3. Tools (PraisonAI-Tools): Specific integrations, external tools (e.g. SurrealDB, Slack), and community extensions MUST go to the `MervinPraison/PraisonAI-Tools` repository. If the feature is a tool, clone `https://github.com/MervinPraison/PraisonAI-Tools`, implement it there, and create the PR using `gh pr create -R MervinPraison/PraisonAI-Tools`.
- Does it duplicate existing functionality? Check if Agent already supports this via existing params (reflection, planning, tools, hooks, memory).
- Does it inherit from Agent properly? New agent types MUST inherit Agent, not wrap it with composition.
- Does it add new dependencies? Only optional deps allowed, must be lazy-imported.
- Will agent.py grow larger? If the change adds >50 lines to agent.py, find a way to extract instead.
- Is there a name collision with existing exports in __init__.py?
If ANY of these conceptual checks fail (excluding routing), add a comment to the issue explaining why and close it. Do NOT create a PR.
STEP 3 — IMPLEMENT:
- Create a new git branch appropriately named (e.g. \`feature/issue-<number>\`)
- Follow protocol-driven design: protocols in core SDK, heavy implementations in wrapper
- Keep changes small and backward-compatible
STEP 4 — TEST:
- Run: cd src/praisonai-agents && PYTHONPATH=. python -m pytest tests/ -x -q --timeout=30
- Ensure no regressions
STEP 5 — CREATE PR:
- Commit with descriptive message, push the branch
- Use the GitHub CLI (\`gh pr create --title "..." --body "Fixes #<number>"\`) to open a Pull Request
CRITICAL: You MUST create the PR automatically using \`gh pr create\`. Do NOT just provide a link.
Respond naturally and helpfully to the user's request using the available tools.
EOF
echo "PROMPT_FILE=/tmp/claude-action/github-context-prompt.txt" >> $GITHUB_ENV
echo "SKIP_EXECUTION=false" >> $GITHUB_ENV
- name: Run Claude Code
if: env.SKIP_EXECUTION != 'true'
shell: bash
run: |
echo "🚀 Running Claude Code with GitHub context..."
# Build command arguments
CMD_ARGS=("-p" "--verbose" "--output-format" "stream-json")
# Add max turns if specified
if [[ -n "${{ inputs.max_turns }}" ]]; then
CMD_ARGS+=("--max-turns" "${{ inputs.max_turns }}")
fi
# Add allowed tools (include GitHub tools by default)
TOOLS="mcp__github__get_issue,mcp__github__get_issue_comments,mcp__github__update_issue,mcp__github__search_issues,mcp__github__list_issues,mcp__github__create_comment,Read,Write,Edit,Bash"
if [[ -n "${{ inputs.allowed_tools }}" ]]; then
TOOLS="$TOOLS,${{ inputs.allowed_tools }}"
fi
CMD_ARGS+=("--allowedTools" "$TOOLS")
# Add disallowed tools
if [[ -n "${{ inputs.disallowed_tools }}" ]]; then
CMD_ARGS+=("--disallowedTools" "${{ inputs.disallowed_tools }}")
fi
echo "📝 Executing Claude Code with prompt from file..."
# Execute Claude Code with timeout, using stdin for the prompt
TIMEOUT_SECONDS=$((${{ inputs.timeout_minutes }} * 60))
timeout $TIMEOUT_SECONDS claude "${CMD_ARGS[@]}" < "${{ env.PROMPT_FILE }}"
echo "✅ Claude Code execution completed"
env:
ANTHROPIC_API_KEY: ${{ inputs.anthropic_api_key }}
CLAUDE_CODE_OAUTH_TOKEN: ${{ inputs.claude_code_oauth_token }}
GITHUB_TOKEN: ${{ inputs.github_token }}
@@ -0,0 +1,91 @@
name: "Claude Issue Triage Action"
description: "Automatically triage GitHub issues using Claude Code"
inputs:
timeout_minutes:
description: "Timeout in minutes for execution"
required: false
default: "5"
claude_code_oauth_token:
description: "Claude Code OAuth token for Pro subscription access"
required: true
github_token:
description: "GitHub token with repo and issues permissions"
required: true
runs:
using: "composite"
steps:
- name: Checkout repository code
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Create prompt
id: generate_prompt
shell: bash
run: |
mkdir -p /tmp/claude-prompts
cat > /tmp/claude-prompts/claude-issue-triage-prompt.txt << 'EOF'
You're an issue triage assistant for GitHub issues. Your task is to analyze the issue and select appropriate labels from the provided list.
IMPORTANT: Don't post any comments or messages to the issue. Your only action should be to apply labels.
Issue Information:
- REPO: ${{ github.repository }}
- ISSUE_NUMBER: ${{ github.event.issue.number }}
TASK OVERVIEW:
1. First, fetch the list of labels available in this repository by running: `gh label list`. Run exactly this command with nothing else.
2. Next, use the GitHub tools to get context about the issue:
- You have access to these tools:
- mcp__github__get_issue: Use this to retrieve the current issue's details including title, description, and existing labels
- mcp__github__get_issue_comments: Use this to read any discussion or additional context provided in the comments
- mcp__github__update_issue: Use this to apply labels to the issue (do not use this for commenting)
- mcp__github__search_issues: Use this to find similar issues that might provide context for proper categorization and to identify potential duplicate issues
- mcp__github__list_issues: Use this to understand patterns in how other issues are labeled
- Start by using mcp__github__get_issue to get the issue details
3. Analyze the issue content, considering:
- The issue title and description
- The type of issue (bug report, feature request, question, etc.)
- Technical areas mentioned
- Severity or priority indicators
- User impact
- Components affected
4. Select appropriate labels from the available labels list provided above:
- Choose labels that accurately reflect the issue's nature
- Be specific but comprehensive
- Select priority labels if you can determine urgency (high-priority, med-priority, or low-priority)
- Consider platform labels (android, ios) if applicable
- If you find similar issues using mcp__github__search_issues, consider using a "duplicate" label if appropriate. Only do so if the issue is a duplicate of another OPEN issue.
5. Apply the selected labels:
- Use mcp__github__update_issue to apply your selected labels
- DO NOT post any comments explaining your decision
- DO NOT communicate directly with users
- If no labels are clearly applicable, do not apply any labels
IMPORTANT GUIDELINES:
- Be thorough in your analysis
- Only select labels from the provided list above
- DO NOT post any comments to the issue
- Your ONLY action should be to apply labels using mcp__github__update_issue
- It's okay to not add any labels if none are clearly applicable
EOF
echo "PROMPT<<DELIMITER" >> $GITHUB_OUTPUT
cat /tmp/claude-prompts/claude-issue-triage-prompt.txt >> $GITHUB_OUTPUT
echo "DELIMITER" >> $GITHUB_OUTPUT
- name: Run Claude Code
uses: ./.github/actions/claude-code-action
with:
direct_prompt: ${{ steps.generate_prompt.outputs.PROMPT }}
allowed_tools: "Bash(gh label list),mcp__github__get_issue,mcp__github__get_issue_comments,mcp__github__update_issue,mcp__github__search_issues,mcp__github__list_issues"
timeout_minutes: ${{ inputs.timeout_minutes }}
claude_code_oauth_token: ${{ inputs.claude_code_oauth_token }}
github_token: ${{ inputs.github_token }}
+11 -3
View File
@@ -251,7 +251,11 @@ jobs:
});
- name: Stash merge gate helpers
run: cp .github/scripts/merge-gate.js /tmp/merge-gate.js
run: |
# Keep scripts together so require('./gate-config.js') works from /tmp
mkdir -p /tmp/merge-gate-scripts /tmp/gh-actions
cp .github/scripts/merge-gate.js .github/scripts/gate-config.js /tmp/merge-gate-scripts/
cp -R .github/actions/claude-code-action /tmp/gh-actions/
- name: Fetch PR branch
env:
@@ -259,6 +263,10 @@ jobs:
run: |
git fetch origin "pull/${PR_NUMBER}/head:pr-${PR_NUMBER}"
git checkout "pr-${PR_NUMBER}"
# PR branches may predate pipeline actions — restore from main checkout
mkdir -p .github/actions
rm -rf .github/actions/claude-code-action
cp -R /tmp/gh-actions/claude-code-action .github/actions/
- name: Claude merge gate assessment (read-only)
id: assess
@@ -315,7 +323,7 @@ jobs:
with:
github-token: ${{ steps.app-token.outputs.token }}
script: |
const mergeGate = require('/tmp/merge-gate.js');
const mergeGate = require('/tmp/merge-gate-scripts/merge-gate.js');
const owner = context.repo.owner;
const repo = context.repo.repo;
const prNumber = ${{ matrix.pr_number }};
@@ -354,7 +362,7 @@ jobs:
with:
github-token: ${{ steps.app-token.outputs.token }}
script: |
const mergeGate = require('/tmp/merge-gate.js');
const mergeGate = require('/tmp/merge-gate-scripts/merge-gate.js');
const comments = await mergeGate.listAllComments(
github, context.repo.owner, context.repo.repo, ${{ matrix.pr_number }}
);
+1 -1
View File
@@ -123,7 +123,7 @@ jobs:
VERSION: ${{ steps.versions.outputs.new_version }}
run: |
TAG="v${VERSION}"
git add pyproject.toml uv.lock
git add pyproject.toml
git commit -m "Release ${TAG}" || exit 0
git tag -f "${TAG}"
git pull --rebase origin main
+4 -2
View File
@@ -638,20 +638,22 @@ air = weather.get_air_quality("Tokyo")
| TTSTool | `OPENAI_API_KEY` or `ELEVENLABS_API_KEY` |
| YouTubeTool | `YOUTUBE_API_KEY` |
| WeatherTool | `OPENWEATHER_API_KEY` |
| SwarmScoreTool | None (uses public API) |
### Using with PraisonAI Agents
```python
from praisonaiagents import Agent
from praisonai_tools import EmailTool, SlackTool, GitHubTool, WeatherTool
from praisonai_tools import EmailTool, SlackTool, GitHubTool, WeatherTool, SwarmScoreTool
agent = Agent(
instructions="You are a helpful assistant with access to email, slack, github, and weather",
instructions="You are a helpful assistant with access to email, slack, github, weather, and trust ratings",
tools=[
EmailTool(provider="gmail"),
SlackTool(),
GitHubTool(),
WeatherTool(),
SwarmScoreTool(),
]
)
+378
View File
@@ -0,0 +1,378 @@
# Inkog Security Analysis Integration
This document describes the integration of [Inkog](https://github.com/inkog-io/inkog) security analysis capabilities into PraisonAI-Tools.
## Overview
Inkog is a specialized security analysis tool for AI agent code that detects vulnerabilities unique to agentic systems. It performs static analysis to catch bugs that only break in agent code, such as:
- **Token bombing** - loops where the LLM controls termination, draining API budgets
- **Recursive tool calling** - one user request fans out into thousands of LLM invocations
- **Prompt injection sinks** - RAG output flowing into system prompts without review
- **Missing oversight** - destructive tools firing without human approval
- **Cross-tenant leakage** - global state shared between agent invocations
- **MCP tool poisoning** - malicious tool descriptions hijacking agents
## Features
### Framework Support
- **21+ frameworks**: LangChain, CrewAI, AutoGen, OpenAI Agents, LangGraph, DSPy, Phidata, and more
- **Multiple languages**: Python, TypeScript, JavaScript, and no-code platforms
- **MCP servers**: Audit Model Context Protocol servers for security issues
### Compliance Mapping
- **EU AI Act** Articles 14/15 (human oversight requirements)
- **NIST AI RMF** (risk management framework)
- **ISO 42001** (AI management systems)
- **OWASP LLM Top 10** (security vulnerabilities)
### Scan Types
- **Core scan**: Fast static analysis with vulnerability detection
- **Deep scan**: AI orchestrator-driven analysis with enhanced context
- **Skill scan**: Security analysis of SKILL.md packages and agent tools
- **MCP scan**: Audit MCP servers for tool poisoning and privilege escalation
## Installation
### Prerequisites
1. **Install Inkog CLI**:
```bash
# macOS (Homebrew)
brew tap inkog-io/inkog && brew install inkog
# Or install via Go
go install github.com/inkog-io/inkog/cmd/inkog@latest
# Or download binary from releases
# https://github.com/inkog-io/inkog/releases
```
2. **Get API Key**:
- Visit [app.inkog.io](https://app.inkog.io) to get a free API key
- Set environment variable: `export INKOG_API_KEY=your_key_here`
3. **Verify Installation**:
```bash
inkog -version
```
## Usage
### Basic Usage
```python
from praisonai_tools import InkogTool
# Initialize the tool
inkog = InkogTool()
# Scan current directory for vulnerabilities
result = inkog.scan_directory(
path=".",
output_format="json",
policy="balanced"
)
# Analyze results
analysis = inkog.analyze_findings(result)
print(analysis)
```
### Standalone Functions
```python
from praisonai_tools.tools.inkog_tool import scan_agent_code, scan_skill_package
# Quick scan using standalone function
result = scan_agent_code("./my-agent", policy="low-noise")
# Scan a skill package
skill_result = scan_skill_package("./my-skill", deep=True)
```
### Advanced Scanning
```python
# Deep scan with AI orchestrator (requires Inkog Deep role)
deep_result = inkog.scan_directory(
path="./agent-code",
output_format="json",
policy="comprehensive",
deep=True,
severity="high"
)
# Generate HTML report
html_report = inkog.scan_directory(
path="./agent-code",
output_format="html",
policy="balanced"
)
# Save report to file
with open("security_report.html", "w") as f:
f.write(html_report)
```
### MCP Server Auditing
```python
# Scan MCP server by name
mcp_result = inkog.mcp_scan("github")
# Scan MCP server with custom repository
mcp_result = inkog.mcp_scan(
"my-server",
repo="https://github.com/org/my-mcp-server",
deep=True
)
```
## Configuration
### Environment Variables
- **`INKOG_API_KEY`** (required): API key from app.inkog.io
- **`INKOG_SERVER_URL`** (optional): Custom server URL for enterprise deployments
### Security Policies
Choose the appropriate policy based on your use case:
- **`low-noise`**: Only exploitable vulnerabilities (good for CI/CD)
- **`balanced`**: Vulnerabilities + risk patterns (default, good for code review)
- **`comprehensive`**: All findings including hardening recommendations (good for security audits)
### Output Formats
- **`json`**: Structured data for programmatic processing
- **`text`**: Human-readable terminal output
- **`html`**: Interactive report with visualizations
- **`sarif`**: SARIF format for GitHub Security tab integration
### Severity Levels
- **`critical`**: Immediate security threats
- **`high`**: Serious vulnerabilities that should be addressed soon
- **`medium`**: Moderate issues worth reviewing
- **`low`**: Minor issues and best practice recommendations
## Integration Examples
### PraisonAI Agent Security Audit
```python
from praisonaiagents import Agent
from praisonai_tools import InkogTool
# Create an agent with security analysis capability
security_agent = Agent(
name="security_auditor",
instructions="You analyze AI agent code for security vulnerabilities",
tools=[InkogTool()]
)
# Audit agent code
audit_result = security_agent.start("""
Scan the ./my-agent directory for security vulnerabilities.
Focus on critical and high severity issues.
Provide a summary of findings and remediation recommendations.
""")
print(audit_result)
```
### CI/CD Pipeline Integration
```python
import sys
from praisonai_tools.tools.inkog_tool import scan_agent_code
def security_check():
"""Security check for CI/CD pipeline."""
result = scan_agent_code(
path=".",
policy="low-noise", # Only critical vulnerabilities
severity="critical"
)
if "error" in result:
print(f"Security scan failed: {result['error']}")
return 1
critical_count = result.get("summary", {}).get("critical", 0)
if critical_count > 0:
print(f"❌ Found {critical_count} critical security issues")
return 1
print("✅ No critical security vulnerabilities found")
return 0
if __name__ == "__main__":
sys.exit(security_check())
```
### Multi-Agent Security Review
```python
from praisonaiagents import Agent, AgentTeam
from praisonai_tools import InkogTool
# Security analysis agent
security_analyzer = Agent(
name="security_analyzer",
instructions="Analyze code for security vulnerabilities using Inkog",
tools=[InkogTool()]
)
# Code reviewer agent
code_reviewer = Agent(
name="code_reviewer",
instructions="Review security findings and provide remediation advice"
)
# Create security review team
security_team = AgentTeam(
name="security_review_team",
agents=[security_analyzer, code_reviewer],
tasks=[
{"agent": "security_analyzer", "task": "Scan the codebase for vulnerabilities"},
{"agent": "code_reviewer", "task": "Review findings and create remediation plan"}
]
)
# Run security review
review_result = security_team.start("Perform comprehensive security review of ./agent-code")
```
## Error Handling
The tool provides detailed error messages for common issues:
```python
result = inkog.scan_directory("./nonexistent")
if "error" in result:
error_msg = result["error"]
if "not installed" in error_msg:
print("Install inkog CLI first")
elif "API_KEY" in error_msg:
print("Set INKOG_API_KEY environment variable")
elif "timeout" in error_msg:
print("Scan timed out - try smaller directory or increase timeout")
else:
print(f"Scan error: {error_msg}")
```
## Best Practices
### 1. Development Workflow
```bash
# Add to your development script
inkog . -policy balanced -output json > security_report.json
```
### 2. Pre-commit Hooks
```yaml
# .pre-commit-config.yaml
repos:
- repo: local
hooks:
- id: inkog-security-scan
name: Inkog Security Scan
entry: inkog
args: [".", "-policy", "low-noise", "-severity", "critical"]
language: system
always_run: true
```
### 3. GitHub Actions
```yaml
name: Security Scan
on: [push, pull_request]
jobs:
security:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Inkog
run: |
curl -fsSL https://inkog.io/install.sh | sh
echo "$HOME/.local/bin" >> $GITHUB_PATH
- name: Run Security Scan
env:
INKOG_API_KEY: ${{ secrets.INKOG_API_KEY }}
run: inkog . -output sarif > results.sarif
- name: Upload SARIF
uses: github/codeql-action/upload-sarif@v2
with:
sarif_file: results.sarif
```
## Privacy & Security
Inkog uses "surgical redaction" to protect sensitive data:
### What Gets Redacted
- AWS access keys, GitHub tokens, Stripe keys
- Database connection strings with passwords
- Private keys and JWT tokens
- High-entropy strings (API keys)
### What is NOT Redacted
- Prompts and templates (needed for injection detection)
- Business logic (needed for flow analysis)
- Configuration values (model names, etc.)
- Normal strings that don't match credential patterns
## Troubleshooting
### Common Issues
1. **"inkog CLI not installed"**
- Install via Homebrew: `brew tap inkog-io/inkog && brew install inkog`
- Or download from releases: https://github.com/inkog-io/inkog/releases
2. **"INKOG_API_KEY required"**
- Get free key at https://app.inkog.io
- Set environment variable: `export INKOG_API_KEY=your_key`
3. **"Deep scan requires Inkog Deep role"**
- Deep scans need special account permissions
- Use standard scan instead: `deep=False`
4. **"Scan timed out"**
- Large codebases may timeout
- Scan specific directories instead of entire project
- Use policy="low-noise" for faster scans
### Debug Mode
Enable verbose output for troubleshooting:
```python
result = inkog.scan_directory(
path=".",
verbose=True # Shows detailed analysis steps
)
```
## Support
- **Documentation**: https://docs.inkog.io
- **GitHub Issues**: https://github.com/inkog-io/inkog/issues
- **Community**: https://discord.gg/NuG4SSGRH
- **Security**: security@inkog.io
## References
- [Inkog Repository](https://github.com/inkog-io/inkog)
- [State of AI Agent Security 2026 Report](https://inkog.io/report)
- [EU AI Act Compliance Guide](https://docs.inkog.io/compliance/eu-ai-act)
- [OWASP LLM Top 10](https://owasp.org/www-project-top-10-for-large-language-model-applications/)
+382
View File
@@ -0,0 +1,382 @@
# NEXUS Prediction Market Integration
This document provides comprehensive information about integrating the NEXUS prediction market API with PraisonAI agents.
## Overview
NEXUS provides live Kalshi/Polymarket prediction market data and arbitrage opportunities through a validated API. The integration allows PraisonAI agents to:
- Access real-time prediction market data
- Identify arbitrage opportunities between markets
- Analyze market sentiment and trends
- Build sophisticated trading and analysis workflows
## API Information
### Endpoints
- **Free Market Data**: `/kalshi?market={market_name}`
- **Paid Arbitrage Analysis**: `/arb/check?markets={market1,market2}` ($0.02 USDC on eip155:8453)
- **Agent Metadata**: `/.well-known/agent.json`
### Validation
The API is fully validated on Agentic Market:
- **Validation URL**: https://agentic.market/validate?url=https://nexus-agent-xa12.onrender.com/arb/check
- **API Card**: https://nexus-agent-xa12.onrender.com/.well-known/agent.json
### Usage Tracking
Include the `X-NEXUS-Source` header to track per-source call volume:
```
X-NEXUS-Source: your-bot-name
```
## Installation
```bash
pip install praisonai-tools
```
## Quick Start
### Basic Usage
```python
from praisonai_tools.tools import NexusPredictionMarketTool
# Initialize the tool
nexus = NexusPredictionMarketTool(source_name="my-agent")
# Get market data
fed_data = nexus.get_market_data("Fed")
print(fed_data)
# Check arbitrage opportunities (paid)
arb_data = nexus.check_arbitrage_opportunities(["Fed", "BTC"])
print(arb_data)
```
### Agent Integration
```python
from praisonaiagents import Agent, tool
from praisonai_tools.tools import nexus_prediction_market_tool
@tool
def get_market_data(market: str) -> str:
"""Get prediction market data."""
return nexus_prediction_market_tool("get_market_data", market=market)
agent = Agent(
name="MarketAnalyst",
tools=[get_market_data],
instructions="Analyze prediction markets using NEXUS data."
)
response = agent.start("What do Fed rate predictions show?")
```
## Features
### Free Features
1. **Market Data Queries**
- Real-time price data
- Volume and liquidity information
- Market sentiment indicators
- Historical trends
2. **Agent Metadata**
- API capabilities and endpoints
- Service status and version info
- Usage guidelines
### Paid Features
1. **Arbitrage Analysis** ($0.02 USDC on eip155:8453)
- Cross-market opportunity detection
- Profit potential calculations
- Risk assessment
- Execution recommendations
## Supported Markets
Common market identifiers include:
- `Fed` - Federal Reserve interest rate predictions
- `BTC` - Bitcoin price predictions
- `Election` - Political election outcomes
- `ETH` - Ethereum price predictions
- `Inflation` - Inflation rate predictions
*Note: Market availability may vary. Check current markets via the API.*
## API Reference
### NexusPredictionMarketTool Class
```python
class NexusPredictionMarketTool:
def __init__(self, source_name: Optional[str] = None, api_key: Optional[str] = None)
def get_market_data(self, market: str) -> Dict[str, Any]
def check_arbitrage_opportunities(self, markets: List[str]) -> Dict[str, Any]
def get_agent_info(self) -> Dict[str, Any]
def run(self, action: str, market: Optional[str] = None, markets: Optional[List[str]] = None) -> str
```
### Function Interface
```python
def nexus_prediction_market_tool(
action: str,
market: Optional[str] = None,
markets: Optional[List[str]] = None,
source_name: Optional[str] = None
) -> str
```
## Examples
### Market Analysis Agent
```python
from praisonaiagents import Agent
from praisonai_tools.tools import NexusPredictionMarketTool
class MarketAnalysisAgent:
def __init__(self):
self.nexus = NexusPredictionMarketTool(source_name="analysis-bot")
def analyze_market_sentiment(self, market):
"""Analyze market sentiment for a specific prediction market."""
data = self.nexus.get_market_data(market)
if data["success"]:
market_data = data["data"]
# Analyze price, volume, trends etc.
return f"Market sentiment for {market}: {self.interpret_data(market_data)}"
else:
return f"Error getting data: {data['error']}"
def interpret_data(self, data):
"""Custom logic to interpret market data."""
# Implementation specific to your analysis needs
pass
```
### Multi-Agent Workflow
```python
from praisonaiagents import Agent, AgentTeam, Task
# Data collector agent
collector = Agent(
name="DataCollector",
tools=[get_market_data],
instructions="Collect prediction market data from NEXUS."
)
# Analyst agent
analyst = Agent(
name="Analyst",
instructions="Analyze market data and provide insights."
)
# Create workflow
team = AgentTeam(agents=[collector, analyst])
# Define tasks
collect_task = Task(
name="collect_data",
description="Collect data for Fed and BTC markets",
agent=collector
)
analyze_task = Task(
name="analyze_trends",
description="Analyze the collected market data",
agent=analyst,
dependencies=[collect_task]
)
# Execute workflow
results = team.run(tasks=[collect_task, analyze_task])
```
## Error Handling
The tool provides comprehensive error handling for common scenarios:
### Network Errors
```python
{
"success": False,
"error": "Failed to fetch market data: Connection timeout",
"market": "Fed"
}
```
### Payment Required (Arbitrage Features)
```python
{
"success": False,
"error": "Payment required for arbitrage analysis. Cost: $0.02 USDC on eip155:8453",
"markets": ["Fed", "BTC"],
"payment_info": "This feature requires payment on eip155:8453 blockchain"
}
```
### Invalid Parameters
```python
{
"success": False,
"error": "Market parameter required for get_market_data action"
}
```
## Best Practices
### 1. Source Identification
Always set a descriptive source name for tracking:
```python
nexus = NexusPredictionMarketTool(source_name="trading-bot-v1")
```
### 2. Error Handling
Always check the `success` field before processing results:
```python
result = nexus.get_market_data("Fed")
if result["success"]:
data = result["data"]
# Process data
else:
print(f"Error: {result['error']}")
```
### 3. Rate Limiting
Implement appropriate delays between requests to avoid overwhelming the API:
```python
import time
import asyncio
async def batch_collect_data(markets):
data = {}
for market in markets:
result = nexus.get_market_data(market)
data[market] = result
await asyncio.sleep(1) # 1 second delay between requests
return data
```
### 4. Caching
Cache results for frequently accessed data:
```python
from functools import lru_cache
import time
@lru_cache(maxsize=100)
def cached_market_data(market, timestamp):
return nexus.get_market_data(market)
def get_cached_data(market, cache_duration=300): # 5 minutes
timestamp = int(time.time() // cache_duration)
return cached_market_data(market, timestamp)
```
## Security Considerations
### 1. API Key Management
If using paid features, securely manage your API keys:
```python
import os
api_key = os.getenv("NEXUS_API_KEY")
nexus = NexusPredictionMarketTool(api_key=api_key)
```
### 2. Input Validation
Validate market identifiers before making requests:
```python
VALID_MARKETS = ["Fed", "BTC", "ETH", "Election", "Inflation"]
def validate_market(market):
if market not in VALID_MARKETS:
raise ValueError(f"Invalid market: {market}")
return market
```
### 3. Rate Limiting
Implement client-side rate limiting to prevent abuse:
```python
from time import time, sleep
class RateLimiter:
def __init__(self, max_calls_per_minute=60):
self.max_calls = max_calls_per_minute
self.calls = []
def wait_if_needed(self):
now = time()
self.calls = [call_time for call_time in self.calls if now - call_time < 60]
if len(self.calls) >= self.max_calls:
sleep(1)
self.calls.append(now)
```
## Troubleshooting
### Common Issues
1. **Connection Timeout**
- Check internet connectivity
- Verify the API endpoint is accessible
- Implement retry logic with exponential backoff
2. **Invalid Market Identifier**
- Verify the market name is correct
- Check available markets via agent info endpoint
- Use exact case-sensitive market names
3. **Payment Required Errors**
- Ensure you have sufficient USDC balance on eip155:8453
- Verify payment setup for arbitrage features
- Contact NEXUS support if payment issues persist
4. **JSON Decode Errors**
- Check API response format
- Verify the endpoint returned valid JSON
- Implement fallback error handling
### Debug Mode
Enable debug logging to troubleshoot issues:
```python
import logging
logging.basicConfig(level=logging.DEBUG)
nexus = NexusPredictionMarketTool(source_name="debug-agent")
# Requests will now be logged with full details
```
## Support and Resources
- **API Documentation**: https://nexus-agent-xa12.onrender.com/.well-known/agent.json
- **Validation Check**: https://agentic.market/validate?url=https://nexus-agent-xa12.onrender.com/arb/check
- **PraisonAI Documentation**: https://docs.praison.ai
- **Issue Tracking**: Report issues via GitHub issues in the PraisonAI-Tools repository
## Contributing
To contribute improvements or report bugs:
1. Fork the PraisonAI-Tools repository
2. Create a feature branch
3. Make your changes
4. Add tests for new functionality
5. Submit a pull request
## License
This tool is part of the PraisonAI-Tools package and follows the same licensing terms.
+357
View File
@@ -0,0 +1,357 @@
# SwarmScore Integration for PraisonAI
SwarmScore is a portable trust rating system for AI agents that provides cryptographically signed reputation scores based on verified execution history. This integration allows PraisonAI agents to leverage trust ratings for enhanced security and decision-making.
## Overview
SwarmScore provides:
- **Portable Trust Ratings**: Cryptographically signed scores that travel with your agent across platforms
- **Verified Execution History**: Trust based on actual performance, not self-reported metrics
- **Consistency Tracking**: 80 jobs at 95% success rate beats 1 job at 100%
- **Agent-to-Agent Discovery**: Machine-readable trust data for autonomous agent interactions
## Installation
```bash
pip install praisonai-tools[swarmscore]
# or
pip install praisonai-tools requests
```
## Quick Start
### Basic Usage
```python
from praisonai_tools.tools import SwarmScoreTool
# Initialize the tool
swarmscore = SwarmScoreTool()
# Load your agent's trust score
result = swarmscore.load_swarmscore("your-agent-slug")
if result.success:
score_data = result.output
print(f"Trust Score: {score_data['score']}")
print(f"Trust Tier: {score_data['tier']}")
print(f"Success Rate: {score_data['success_rate']}%")
else:
print(f"Error: {result.error}")
```
### Standalone Functions
```python
from praisonai_tools.tools import (
load_swarmscore_by_slug,
verify_swarmscore_freshness,
get_agent_discovery_manifest
)
# Load score data
try:
score_data = load_swarmscore_by_slug("your-agent-slug")
print(f"Score: {score_data['score']}")
except Exception as e:
print(f"Failed to load score: {e}")
# Verify score freshness
try:
verification = verify_swarmscore_freshness(score_data['verify_payload'])
print(f"Verification: {verification}")
except Exception as e:
print(f"Verification failed: {e}")
# Get discovery manifest
try:
manifest = get_agent_discovery_manifest()
print(f"Discovery data: {manifest}")
except Exception as e:
print(f"Failed to get manifest: {e}")
```
## Integration Patterns
### 1. Trust-Based Decision Making
```python
from praisonai_tools.tools import SwarmScoreTool
def execute_with_trust_level(agent_slug, task):
"""Execute task based on agent's trust level."""
swarmscore = SwarmScoreTool()
result = swarmscore.load_swarmscore(agent_slug)
if not result.success:
# Default to restricted mode if score unavailable
return execute_restricted(task)
trust_score = result.output.get('score', 0)
if trust_score >= 80:
# High trust - full autonomy
return execute_autonomous(task)
elif trust_score >= 60:
# Medium trust - human oversight
return execute_supervised(task)
else:
# Low trust - restricted operations
return execute_restricted(task)
```
### 2. Agent Authentication
```python
def authenticate_agent(agent_slug):
"""Verify agent identity and trust level."""
swarmscore = SwarmScoreTool()
# Load and verify score
load_result = swarmscore.load_swarmscore(agent_slug)
if not load_result.success:
return False, "Failed to load trust score"
# Verify freshness
verify_payload = load_result.output.get('verify_payload')
if verify_payload:
verify_result = swarmscore.verify_swarmscore(verify_payload)
if not verify_result.success:
return False, "Trust score verification failed"
trust_score = load_result.output.get('score', 0)
return trust_score >= 50, f"Trust score: {trust_score}"
```
### 3. Multi-Agent Coordination
```python
def find_trusted_collaborators(task_requirements):
"""Find agents suitable for collaboration based on trust scores."""
swarmscore = SwarmScoreTool()
# Get discovery manifest for agent lookup
manifest_result = swarmscore.get_discovery_manifest()
if not manifest_result.success:
return []
trusted_agents = []
discovery_data = manifest_result.output
# Example: Filter agents by trust score and capabilities
for agent_id in discovery_data.get('agents', []):
score_result = swarmscore.load_swarmscore(agent_id)
if score_result.success:
score_data = score_result.output
if score_data.get('score', 0) >= 70:
trusted_agents.append({
'id': agent_id,
'score': score_data['score'],
'capabilities': score_data.get('capabilities', [])
})
return trusted_agents
```
## PraisonAI Agent Integration
### Using with PraisonAI Agent Tools
```python
from praisonaiagents import Agent, tool
from praisonai_tools.tools import SwarmScoreTool
@tool
def check_agent_trust_score(agent_slug: str) -> str:
"""Check the trust score for an agent."""
swarmscore = SwarmScoreTool()
result = swarmscore.load_swarmscore(agent_slug)
if result.success:
score_data = result.output
return f"Agent {agent_slug} has trust score {score_data.get('score', 'unknown')} with {score_data.get('success_rate', 'unknown')}% success rate"
else:
return f"Failed to load trust score for {agent_slug}: {result.error}"
# Create agent with SwarmScore capability
agent = Agent(
name="trust-aware-agent",
instructions="You can check agent trust scores using the check_agent_trust_score tool.",
tools=[check_agent_trust_score]
)
# Use the agent
response = agent.start("What is the trust score for agent 'trading-bot-v2'?")
print(response)
```
### Agent with Built-in Trust Management
```python
from praisonaiagents import Agent
from praisonai_tools.tools import SwarmScoreTool
class TrustAwareAgent(Agent):
"""PraisonAI Agent with built-in SwarmScore integration."""
def __init__(self, agent_slug: str, **kwargs):
super().__init__(**kwargs)
self.agent_slug = agent_slug
self.swarmscore = SwarmScoreTool()
self._trust_score = None
async def check_trust_level(self):
"""Check and cache current trust level."""
result = self.swarmscore.load_swarmscore(self.agent_slug)
if result.success:
self._trust_score = result.output.get('score', 0)
return self._trust_score
def can_execute_action(self, action_type: str) -> bool:
"""Determine if action can be executed based on trust level."""
if self._trust_score is None:
return False # Default to safe
# Define trust requirements for different actions
trust_requirements = {
'read': 30,
'write': 60,
'execute': 80,
'admin': 95
}
required_trust = trust_requirements.get(action_type, 100)
return self._trust_score >= required_trust
# Usage
agent = TrustAwareAgent(
agent_slug="my-agent-123",
name="secure-agent",
instructions="Execute tasks based on trust level"
)
# Check trust before executing
trust_score = await agent.check_trust_level()
if agent.can_execute_action('execute'):
print("Proceeding with high-trust execution")
else:
print("Restricted to low-trust operations")
```
## API Reference
### SwarmScoreTool Class
#### Methods
**`__init__(api_base_url: str = "https://api.swarmsync.ai/v1/swarmscore/")`**
- Initialize SwarmScore tool
- `api_base_url`: Base URL for SwarmScore API
**`load_swarmscore(slug: str) -> ToolResult`**
- Load SwarmScore data by agent slug
- Returns: ToolResult with passport, certificate, and verification data
**`verify_swarmscore(verify_payload: Dict[str, Any]) -> ToolResult`**
- Verify SwarmScore freshness using verification payload
- Returns: ToolResult with verification status
**`get_discovery_manifest() -> ToolResult`**
- Get machine-readable agent discovery manifest
- Returns: ToolResult with discovery data
**`run(action: str, **kwargs) -> ToolResult`**
- Execute SwarmScore action
- Actions: 'load', 'verify', 'discover'
### Standalone Functions
**`load_swarmscore_by_slug(slug: str) -> Dict[str, Any]`**
- Load SwarmScore data by agent slug
- Raises: Exception if loading fails
**`verify_swarmscore_freshness(verify_payload: Dict[str, Any]) -> Dict[str, Any]`**
- Verify SwarmScore freshness
- Raises: Exception if verification fails
**`get_agent_discovery_manifest() -> Dict[str, Any]`**
- Get agent discovery manifest
- Raises: Exception if retrieval fails
## SwarmScore Data Structure
```json
{
"score": 85,
"tier": "gold",
"jobs_completed": 150,
"success_rate": 94.5,
"consistency_score": 88,
"signed_certificate": "...",
"verify_payload": {
"signature": "...",
"timestamp": "...",
"agent_id": "..."
},
"capabilities": ["trading", "analysis", "reporting"],
"last_updated": "2024-01-15T10:30:00Z"
}
```
## Error Handling
```python
from praisonai_tools.tools import SwarmScoreTool
swarmscore = SwarmScoreTool()
result = swarmscore.load_swarmscore("agent-123")
if not result.success:
error_msg = result.error
if "network" in error_msg.lower():
# Handle network issues
print("Network connectivity problem")
elif "not found" in error_msg.lower():
# Handle missing agent
print("Agent not registered in SwarmScore")
else:
# Handle other errors
print(f"Unexpected error: {error_msg}")
```
## Best Practices
1. **Cache Trust Scores**: Cache scores temporarily to avoid excessive API calls
2. **Graceful Degradation**: Have fallback behavior when SwarmScore is unavailable
3. **Verify Freshness**: Always verify score freshness for critical decisions
4. **Trust Thresholds**: Define clear trust score requirements for different operations
5. **Error Handling**: Handle network and API errors gracefully
## Getting Started with SwarmScore
1. **Register**: Create an account at [SwarmSync.AI](https://swarmsync.ai)
2. **Get Your Slug**: Obtain your agent identifier from the dashboard
3. **Start Building**: Begin tracking execution history to build trust
4. **Integrate**: Use this tool to incorporate trust ratings into your workflows
## Resources
- [SwarmScore Documentation](https://swarmsync.ai/docs/protocol-specs/swarmscore)
- [SwarmScore GitHub Spec](https://github.com/swarmsync-ai/swarmscore-spec)
- [SwarmSync.AI Platform](https://swarmsync.ai)
- [Example Implementation](../examples/swarmscore_example.py)
## Contributing
To contribute improvements to the SwarmScore integration:
1. Fork the PraisonAI-Tools repository
2. Create a feature branch
3. Make your changes
4. Add tests for new functionality
5. Submit a pull request
## License
This integration is part of PraisonAI-Tools and follows the same license terms.
+261
View File
@@ -0,0 +1,261 @@
"""
Inkog Security Analysis Tool Example
This example demonstrates how to use the Inkog tool to analyze AI agent code
for security vulnerabilities like token bombing, prompt injection, and more.
Prerequisites:
1. Install inkog CLI: brew tap inkog-io/inkog && brew install inkog
2. Get free API key from https://app.inkog.io
3. Set environment variable: export INKOG_API_KEY=your_key_here
The Inkog tool can detect:
- Token bombing attacks (loops where LLM controls termination)
- Prompt injection vulnerabilities
- Recursive tool calling without cycle detection
- Missing human oversight for destructive operations
- Cross-tenant data leakage
- MCP tool poisoning
- EU AI Act / NIST / OWASP compliance issues
"""
import os
from praisonai_tools import InkogTool, scan_agent_code
def basic_security_scan():
"""Perform a basic security scan of the current directory."""
print("🔍 Basic Security Scan Example")
print("=" * 50)
# Method 1: Using the class
inkog = InkogTool()
result = inkog.scan_directory(
path=".",
output_format="json",
policy="balanced" # balanced, low-noise, comprehensive
)
if "error" in result:
print(f"❌ Scan failed: {result['error']}")
return
# Analyze the results
analysis = inkog.analyze_findings(result)
print(analysis)
print("\n" + "=" * 50)
def advanced_security_scan():
"""Perform an advanced security scan with deep analysis."""
print("🔬 Advanced Security Scan Example (Deep Analysis)")
print("=" * 60)
inkog = InkogTool()
# Deep scan provides enhanced analysis with AI orchestrator
result = inkog.scan_directory(
path=".",
output_format="json",
policy="comprehensive", # Show all findings including best practices
deep=True, # Enable AI orchestrator analysis (requires Inkog Deep role)
verbose=True
)
if "error" in result:
print(f"❌ Deep scan failed: {result['error']}")
print("Note: Deep scans require the Inkog Deep role on your account")
return
# Show detailed analysis
analysis = inkog.analyze_findings(result)
print(analysis)
# Show raw findings for debugging
if "server_findings" in result:
print(f"\n📊 Raw Findings ({len(result['server_findings'])} total):")
for i, finding in enumerate(result["server_findings"][:3]): # Show first 3
print(f"{i+1}. {finding.get('pattern', 'Unknown')} - {finding.get('severity', 'N/A')}")
print(f" File: {finding.get('file', 'N/A')}:{finding.get('line', 'N/A')}")
print(f" Message: {finding.get('message', 'N/A')}")
print()
print("=" * 60)
def skill_package_scan():
"""Scan a SKILL.md package or MCP server."""
print("📦 Skill Package Security Scan Example")
print("=" * 50)
inkog = InkogTool()
# Scan current directory as a skill package
result = inkog.skill_scan(
path=".",
output_format="json",
deep=False
)
if "error" in result:
print(f"❌ Skill scan failed: {result['error']}")
return
analysis = inkog.analyze_findings(result)
print(analysis)
print("\n" + "=" * 50)
def mcp_server_scan():
"""Scan an MCP server from the registry."""
print("🔌 MCP Server Security Scan Example")
print("=" * 50)
inkog = InkogTool()
# Scan a popular MCP server (GitHub)
result = inkog.mcp_scan(
server_name="github",
output_format="json"
)
if "error" in result:
print(f"❌ MCP scan failed: {result['error']}")
return
analysis = inkog.analyze_findings(result)
print(analysis)
print("\n" + "=" * 50)
def standalone_function_example():
"""Example using standalone functions for quick scans."""
print("⚡ Standalone Function Examples")
print("=" * 40)
# Quick scan using standalone function
print("Running quick security scan...")
result = scan_agent_code(
path=".",
output_format="json",
policy="low-noise", # Only critical vulnerabilities
deep=False
)
if isinstance(result, dict) and "error" not in result:
tool = InkogTool()
analysis = tool.analyze_findings(result)
print(analysis)
else:
print(f"Scan result: {result}")
print("\n" + "=" * 40)
def html_report_example():
"""Generate an HTML security report."""
print("📄 HTML Report Generation Example")
print("=" * 45)
inkog = InkogTool()
# Generate HTML report
html_report = inkog.scan_directory(
path=".",
output_format="html",
policy="comprehensive",
deep=False
)
if isinstance(html_report, str) and not html_report.startswith("Error"):
# Save report to file
report_path = "inkog_security_report.html"
with open(report_path, "w", encoding="utf-8") as f:
f.write(html_report)
print(f"✅ HTML report generated: {report_path}")
print(f"📄 Report size: {len(html_report)} characters")
else:
print(f"❌ HTML report failed: {html_report}")
print("=" * 45)
def compliance_scan_example():
"""Example focused on compliance mapping."""
print("🏛️ Compliance Mapping Example")
print("=" * 40)
# Use comprehensive policy to see compliance mappings
inkog = InkogTool()
result = inkog.scan_directory(
path=".",
output_format="json",
policy="comprehensive", # Shows compliance mappings
deep=False
)
if "error" in result:
print(f"❌ Compliance scan failed: {result['error']}")
return
# Look for compliance information
findings = result.get("server_findings", [])
compliance_findings = []
for finding in findings:
if any(term in finding.get("message", "").lower()
for term in ["eu ai act", "nist", "owasp", "iso", "article"]):
compliance_findings.append(finding)
if compliance_findings:
print(f"📋 Found {len(compliance_findings)} compliance-related findings:")
for finding in compliance_findings[:3]: # Show first 3
print(f"• {finding.get('pattern', 'Unknown')}")
print(f" {finding.get('message', 'N/A')[:100]}...")
else:
print("✅ No specific compliance violations found")
print("=" * 40)
def main():
"""Run all examples."""
print("Inkog Security Analysis Tool Examples")
print("=" * 60)
# Check if inkog is available
api_key = os.environ.get("INKOG_API_KEY")
if not api_key:
print("⚠️ WARNING: INKOG_API_KEY environment variable not set")
print(" Get your free API key at https://app.inkog.io")
print(" Set it with: export INKOG_API_KEY=your_key_here")
print()
try:
# Run examples in order of complexity
basic_security_scan()
standalone_function_example()
skill_package_scan()
html_report_example()
compliance_scan_example()
# Advanced features (may require special access)
print("\n🎓 Advanced Examples (may require Inkog Deep role):")
print("-" * 50)
advanced_security_scan()
mcp_server_scan()
except Exception as e:
print(f"❌ Example failed: {e}")
print("This may be due to:")
print("1. Inkog CLI not installed")
print("2. Missing INKOG_API_KEY environment variable")
print("3. Network connectivity issues")
print("4. Insufficient permissions for deep scans")
if __name__ == "__main__":
main()
+198
View File
@@ -0,0 +1,198 @@
"""Example: Using NEXUS Prediction Market Tool with PraisonAI
This example demonstrates how to use the NEXUS prediction market tool to:
1. Get free market data for prediction markets
2. Check arbitrage opportunities between markets (paid feature)
3. Build an intelligent prediction market analysis agent
NEXUS API Information:
- Free endpoint: /kalshi?market={market_name}
- Paid endpoint: /arb/check?markets={market1,market2} ($0.02 USDC on eip155:8453)
- Validation: https://agentic.market/validate?url=https://nexus-agent-xa12.onrender.com/arb/check
"""
from praisonaiagents import Agent, tool
from praisonai_tools.tools.nexus_prediction_market_tool import NexusPredictionMarketTool
import json
# Example 1: Direct tool usage
def example_direct_usage():
"""Example of using the NEXUS tool directly."""
print("=== Direct NEXUS Tool Usage ===\n")
# Initialize the tool
nexus_tool = NexusPredictionMarketTool(source_name="praisonai-example")
# Get market data for Fed predictions
print("1. Getting Fed market data...")
fed_data = nexus_tool.get_market_data("Fed")
print(json.dumps(fed_data, indent=2))
print()
# Get BTC market data
print("2. Getting BTC market data...")
btc_data = nexus_tool.get_market_data("BTC")
print(json.dumps(btc_data, indent=2))
print()
# Check arbitrage opportunities (paid feature)
print("3. Checking arbitrage opportunities between Fed and BTC markets...")
arb_data = nexus_tool.check_arbitrage_opportunities(["Fed", "BTC"])
print(json.dumps(arb_data, indent=2))
print()
# Get agent information
print("4. Getting NEXUS agent information...")
agent_info = nexus_tool.get_agent_info()
print(json.dumps(agent_info, indent=2))
# Example 2: Create a custom tool wrapper for agents
@tool
def get_prediction_market_data(market: str) -> str:
"""Get prediction market data for a specific market.
Args:
market: Market identifier (e.g., 'Fed', 'BTC', 'Election')
Returns:
JSON string with market data including prices and trends
"""
nexus_tool = NexusPredictionMarketTool(source_name="praisonai-agent")
return nexus_tool.run(action="get_market_data", market=market)
@tool
def check_prediction_market_arbitrage(markets: str) -> str:
"""Check for arbitrage opportunities between prediction markets.
Args:
markets: Comma-separated list of markets to analyze (e.g., 'Fed,BTC')
Returns:
JSON string with arbitrage opportunities and analysis
Note:
This is a paid feature requiring $0.02 USDC on eip155:8453
"""
markets_list = [m.strip() for m in markets.split(',')]
nexus_tool = NexusPredictionMarketTool(source_name="praisonai-agent")
return nexus_tool.run(action="check_arbitrage", markets=markets_list)
# Example 3: Prediction market analysis agent
def example_prediction_market_agent():
"""Example of a prediction market analysis agent."""
print("\n=== Prediction Market Analysis Agent ===\n")
# Create an agent with NEXUS prediction market tools
market_agent = Agent(
name="PredictionMarketAnalyst",
instructions="""You are an expert prediction market analyst.
Your capabilities:
- Analyze prediction market data from Kalshi and Polymarket via NEXUS API
- Identify market trends and sentiment
- Spot potential arbitrage opportunities
- Provide insights on market dynamics
When analyzing markets:
1. First get current market data
2. Analyze the data for trends and insights
3. Look for arbitrage opportunities if requested
4. Provide clear explanations of your findings
Be thorough in your analysis and cite specific data points.""",
tools=[get_prediction_market_data, check_prediction_market_arbitrage],
llm="gpt-4o-mini"
)
# Example queries
queries = [
"Get the current Fed interest rate prediction market data and analyze what it shows about market sentiment.",
"Compare Fed and BTC prediction markets. What insights can you derive?",
"Check if there are any arbitrage opportunities between Fed and BTC markets."
]
for i, query in enumerate(queries, 1):
print(f"Query {i}: {query}")
response = market_agent.start(query)
print(f"Response: {response}\n")
print("-" * 80)
# Example 4: Multi-agent prediction market workflow
def example_multi_agent_workflow():
"""Example of a multi-agent workflow for comprehensive market analysis."""
print("\n=== Multi-Agent Prediction Market Workflow ===\n")
# Data gathering agent
data_agent = Agent(
name="MarketDataCollector",
instructions="""You collect prediction market data from NEXUS API.
Focus on gathering comprehensive data for analysis.""",
tools=[get_prediction_market_data]
)
# Analysis agent
analysis_agent = Agent(
name="MarketAnalyst",
instructions="""You analyze prediction market data to identify trends,
sentiment, and trading opportunities. Provide detailed insights.""",
tools=[]
)
# Arbitrage agent
arbitrage_agent = Agent(
name="ArbitrageSpecialist",
instructions="""You specialize in finding arbitrage opportunities
between prediction markets. Assess risk and profitability.""",
tools=[check_prediction_market_arbitrage]
)
# Example workflow
markets = ["Fed", "BTC", "Election"]
print("Step 1: Collecting market data...")
market_data = {}
for market in markets:
print(f"Collecting data for {market} market...")
data = data_agent.start(f"Get current market data for {market}")
market_data[market] = data
print(f"Data collected for {market}: {len(data)} characters")
print("\nStep 2: Analyzing market trends...")
combined_data = "\n".join([f"{market}: {data}" for market, data in market_data.items()])
analysis = analysis_agent.start(f"Analyze this prediction market data and identify key trends:\n{combined_data}")
print(f"Analysis: {analysis[:200]}...")
print("\nStep 3: Checking arbitrage opportunities...")
arb_analysis = arbitrage_agent.start("Check for arbitrage opportunities between Fed,BTC markets")
print(f"Arbitrage analysis: {arb_analysis[:200]}...")
if __name__ == "__main__":
print("NEXUS Prediction Market Tool Examples")
print("=" * 50)
# Run examples
try:
example_direct_usage()
example_prediction_market_agent()
example_multi_agent_workflow()
except Exception as e:
print(f"Error running examples: {e}")
print("\nNote: Some features require API access or payment.")
print("Free features: Market data queries")
print("Paid features: Arbitrage analysis ($0.02 USDC on eip155:8453)")
print("\n" + "=" * 50)
print("Examples completed!")
print("\nTo use in your own projects:")
print("1. Install: pip install praisonai-tools")
print("2. Import: from praisonai_tools.tools import NexusPredictionMarketTool")
print("3. Use: tool = NexusPredictionMarketTool()")
print("\nFor arbitrage features, ensure you have the required payment setup.")
+141
View File
@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""
SwarmScore Integration Example for PraisonAI
This example demonstrates how to integrate SwarmScore trust ratings
with PraisonAI agents for portable reputation management.
SwarmScore provides cryptographically signed trust ratings based on
verified execution history, success rates, and consistency.
Requirements:
pip install requests praisonai-tools
Usage:
python swarmscore_example.py
"""
import json
from praisonai_tools.tools import SwarmScoreTool, load_swarmscore_by_slug
def main():
"""Demonstrate SwarmScore integration with PraisonAI."""
print("🔍 SwarmScore Integration Example")
print("=" * 50)
# Initialize SwarmScore tool
swarmscore = SwarmScoreTool()
# Example agent slug (replace with actual agent identifier)
agent_slug = "example-agent-123"
print(f"\n📊 Loading SwarmScore for agent: {agent_slug}")
# Load SwarmScore data
result = swarmscore.load_swarmscore(agent_slug)
if result.success:
print("✅ SwarmScore loaded successfully!")
# Display basic score information
score_data = result.output
if 'score' in score_data:
print(f" Trust Score: {score_data['score']}")
if 'tier' in score_data:
print(f" Trust Tier: {score_data['tier']}")
if 'jobs_completed' in score_data:
print(f" Jobs Completed: {score_data['jobs_completed']}")
if 'success_rate' in score_data:
print(f" Success Rate: {score_data['success_rate']}%")
print("\n🔍 Full SwarmScore data:")
print(json.dumps(score_data, indent=2))
# Verify score freshness
if 'verify_payload' in score_data:
print("\n🔐 Verifying score freshness...")
verify_result = swarmscore.verify_swarmscore(score_data['verify_payload'])
if verify_result.success:
print("✅ Score verification successful!")
print(f" Verification data: {verify_result.output}")
else:
print(f"❌ Score verification failed: {verify_result.error}")
else:
print(f"❌ Failed to load SwarmScore: {result.error}")
print("\n💡 This might happen if:")
print(" • The agent slug doesn't exist in SwarmScore")
print(" • The SwarmScore API is temporarily unavailable")
print(" • Network connectivity issues")
# Get discovery manifest
print("\n🌐 Loading agent discovery manifest...")
manifest_result = swarmscore.get_discovery_manifest()
if manifest_result.success:
print("✅ Discovery manifest loaded!")
print(" This contains machine-readable data for agent-to-agent discovery")
print(f" Manifest keys: {list(manifest_result.output.keys())}")
else:
print(f"❌ Failed to load manifest: {manifest_result.error}")
# Demonstrate standalone function usage
print("\n🛠️ Using standalone functions:")
try:
# This will likely fail with the example slug, but shows the API
score_data = load_swarmscore_by_slug(agent_slug)
print(f"✅ Standalone function succeeded: {score_data}")
except Exception as e:
print(f"❌ Standalone function failed (expected): {e}")
print("\n📚 Next Steps:")
print(" 1. Register your agent at https://swarmsync.ai")
print(" 2. Get your agent slug from the SwarmSync dashboard")
print(" 3. Replace 'example-agent-123' with your actual slug")
print(" 4. Integrate SwarmScore checks into your agent workflows")
print("\n 📖 Documentation: https://swarmsync.ai/docs/protocol-specs/swarmscore")
def agent_workflow_example():
"""Example of how to integrate SwarmScore into an agent workflow."""
print("\n🤖 Agent Workflow Integration Example")
print("=" * 45)
# Simulated agent execution flow
agent_id = "my-trading-agent"
# 1. Load current trust score before executing task
swarmscore = SwarmScoreTool()
score_result = swarmscore.load_swarmscore(agent_id)
if score_result.success:
trust_score = score_result.output.get('score', 0)
print(f"🔍 Agent {agent_id} current trust score: {trust_score}")
# 2. Make decisions based on trust level
if trust_score >= 80:
print("✅ High trust - proceeding with full autonomy")
task_approach = "autonomous"
elif trust_score >= 60:
print("⚠️ Medium trust - requesting human approval for critical actions")
task_approach = "supervised"
else:
print("❌ Low trust - operating in restricted mode")
task_approach = "restricted"
# 3. Execute task based on trust level
print(f"🚀 Executing task with {task_approach} approach...")
# 4. After task completion, the score would be updated by SwarmSync
# (This happens externally based on verified execution results)
print("📝 Task completed - SwarmScore will be updated based on results")
else:
print(f"⚠️ Could not load trust score: {score_result.error}")
print("🔄 Falling back to default security settings")
if __name__ == "__main__":
main()
agent_workflow_example()
+26
View File
@@ -104,6 +104,11 @@ def __getattr__(name):
# ArXiv
"ArxivTool": "arxiv_tool",
"arxiv_search": "arxiv_tool",
# SwarmScore
"SwarmScoreTool": "swarmscore_tool",
"load_swarmscore_by_slug": "swarmscore_tool",
"verify_swarmscore_freshness": "swarmscore_tool",
"get_agent_discovery_manifest": "swarmscore_tool",
# Calculator
"CalculatorTool": "calculator_tool",
"calculate": "calculator_tool",
@@ -545,6 +550,14 @@ def __getattr__(name):
"N8nWorkflowTool": "praisonai_tools.n8n.n8n_workflow",
"n8n_workflow": "praisonai_tools.n8n.n8n_workflow",
"n8n_list_workflows": "praisonai_tools.n8n.n8n_workflow",
# NEXUS Prediction Market Tool
"NexusPredictionMarketTool": "nexus_prediction_market_tool",
"nexus_prediction_market_tool": "nexus_prediction_market_tool",
# Inkog Security Tool
"InkogTool": "inkog_tool",
"scan_agent_code": "inkog_tool",
"scan_skill_package": "inkog_tool",
"scan_mcp_server": "inkog_tool",
}
if name in tool_map:
@@ -654,6 +667,11 @@ __all__ = [
# ArXiv Tool
"ArxivTool",
"arxiv_search",
# SwarmScore Tool
"SwarmScoreTool",
"load_swarmscore_by_slug",
"verify_swarmscore_freshness",
"get_agent_discovery_manifest",
# Calculator Tool
"CalculatorTool",
"calculate",
@@ -895,4 +913,12 @@ __all__ = [
"LangExtractTool",
"langextract_extract",
"langextract_render_file",
# NEXUS Prediction Market Tool
"NexusPredictionMarketTool",
"nexus_prediction_market_tool",
# Inkog Security Tool
"InkogTool",
"scan_agent_code",
"scan_skill_package",
"scan_mcp_server",
]
+569
View File
@@ -0,0 +1,569 @@
"""Inkog Security Analysis Tool for PraisonAI Agents.
Security analysis for AI agent code using Inkog.
Detects token bombing, prompt injection, missing oversight, and compliance gaps.
Usage:
from praisonai_tools import InkogTool
inkog = InkogTool()
results = inkog.scan_directory("./my-agent")
# Or use standalone functions
from praisonai_tools.tools.inkog_tool import scan_agent_code
results = scan_agent_code("./my-agent")
Features:
- Static analysis of AI agent code
- Token bombing detection
- Prompt injection vulnerability scanning
- Recursive tool calling detection
- EU AI Act / NIST / OWASP compliance mapping
- Support for 21+ frameworks (LangChain, CrewAI, AutoGen, etc.)
- Multiple output formats (text, JSON, HTML, SARIF)
Environment Variables:
INKOG_API_KEY: API key from app.inkog.io (required)
INKOG_SERVER_URL: Custom server URL (optional)
"""
import os
import json
import logging
import subprocess
from typing import Any, Dict, Optional, Union
from praisonai_tools.tools.base import BaseTool
logger = logging.getLogger(__name__)
def _check_inkog_available(api_key: Optional[str] = None) -> tuple[bool, Optional[str]]:
"""Check if inkog CLI is available and properly configured.
Args:
api_key: Optional API key to validate. Falls back to the
INKOG_API_KEY environment variable when not provided.
Returns:
Tuple of (is_available, error_message)
"""
# Check if inkog CLI is installed
try:
result = subprocess.run(
["inkog", "-version"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode != 0:
return False, "inkog CLI is not properly installed. Install with: brew tap inkog-io/inkog && brew install inkog"
except FileNotFoundError:
return False, "inkog CLI is not installed. Install with: brew tap inkog-io/inkog && brew install inkog"
except subprocess.TimeoutExpired:
return False, "inkog CLI is not responding. Please check installation."
except Exception as e:
return False, f"Error checking inkog installation: {e}"
# Check for API key (instance-provided key takes precedence over env var)
api_key = api_key or os.environ.get("INKOG_API_KEY")
if not api_key:
return False, "INKOG_API_KEY environment variable is required. Get your free key at https://app.inkog.io"
return True, None
class InkogTool(BaseTool):
"""Comprehensive security analysis tool for AI agent code using Inkog.
Inkog provides static analysis specifically designed for AI agents, detecting:
- Token bombing attacks (loops where LLM controls termination)
- Prompt injection vulnerabilities
- Recursive tool calling without cycle detection
- Missing human oversight for destructive operations
- Cross-tenant data leakage
- MCP tool poisoning
Supports 21+ frameworks including LangChain, CrewAI, AutoGen, OpenAI Agents,
and provides compliance mapping to EU AI Act, NIST AI RMF, and OWASP LLM Top 10.
Requires:
- inkog CLI: brew tap inkog-io/inkog && brew install inkog
- INKOG_API_KEY environment variable (free at app.inkog.io)
"""
name = "inkog_security"
description = "Analyze AI agent code for security vulnerabilities using Inkog static analysis."
def __init__(self, api_key: Optional[str] = None, server_url: Optional[str] = None):
"""Initialize InkogTool.
Args:
api_key: Optional API key. If not provided, uses INKOG_API_KEY env var.
server_url: Optional server URL. If not provided, uses INKOG_SERVER_URL env var.
"""
super().__init__()
self._api_key = api_key
self._server_url = server_url
def _get_api_key(self) -> Optional[str]:
"""Get API key from instance or environment."""
return self._api_key or os.environ.get("INKOG_API_KEY")
def _get_server_url(self) -> Optional[str]:
"""Get server URL from instance or environment."""
return self._server_url or os.environ.get("INKOG_SERVER_URL")
def run(
self,
action: str = "scan",
path: str = ".",
output_format: str = "json",
policy: str = "balanced",
severity: str = "low",
deep: bool = False,
**kwargs
) -> Union[str, Dict[str, Any]]:
"""Execute inkog security analysis.
Args:
action: Action to perform ("scan", "skill-scan", "mcp-scan")
path: Path to scan (directory or file)
output_format: Output format ("json", "text", "html", "sarif")
policy: Security policy ("low-noise", "balanced", "comprehensive")
severity: Minimum severity ("critical", "high", "medium", "low")
deep: Enable deep scan with AI orchestrator analysis
**kwargs: Additional arguments
Returns:
Scan results as dict (JSON) or formatted string (text/html)
"""
action = action.lower().replace("-", "_")
if action == "scan":
return self.scan_directory(
path=path,
output_format=output_format,
policy=policy,
severity=severity,
deep=deep
)
elif action == "skill_scan":
return self.skill_scan(
path=path,
output_format=output_format,
deep=deep,
repo=kwargs.get("repo")
)
elif action == "mcp_scan":
return self.mcp_scan(
server_name=kwargs.get("server_name", path),
output_format=output_format,
deep=deep,
repo=kwargs.get("repo")
)
else:
return {"error": f"Unknown action: {action}"}
def scan_directory(
self,
path: str = ".",
output_format: str = "json",
policy: str = "balanced",
severity: str = "low",
deep: bool = False,
verbose: bool = False
) -> Union[Dict[str, Any], str]:
"""Scan a directory or file for AI agent security vulnerabilities.
Args:
path: Path to scan (directory or file)
output_format: Output format ("json", "text", "html", "sarif")
policy: Security policy ("low-noise", "balanced", "comprehensive")
severity: Minimum severity level ("critical", "high", "medium", "low")
deep: Enable deep scan with AI orchestrator analysis
verbose: Enable verbose output
Returns:
Scan results as dict (JSON format) or string (other formats)
"""
is_available, error = _check_inkog_available(self._get_api_key())
if not is_available:
logger.error(error)
return {"error": error}
try:
# Build command
cmd = ["inkog"]
if deep:
cmd.append("-deep")
cmd.extend([
"-output", output_format,
"-policy", policy,
"-severity", severity,
])
if verbose:
cmd.append("-verbose")
# Add server URL if specified
server_url = self._get_server_url()
if server_url:
cmd.extend(["-server", server_url])
# Add path
cmd.append(path)
# Set environment
env = os.environ.copy()
api_key = self._get_api_key()
if api_key:
env["INKOG_API_KEY"] = api_key
# Execute scan
result = subprocess.run(
cmd,
capture_output=True,
text=True,
env=env,
timeout=300 # 5 minutes timeout
)
if result.returncode != 0:
error_msg = f"Inkog scan failed: {result.stderr or result.stdout}"
logger.error(error_msg)
return {"error": error_msg}
# Parse output based on format
if output_format == "json":
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
return {"error": f"Failed to parse JSON output: {e}", "raw_output": result.stdout}
else:
return result.stdout
except subprocess.TimeoutExpired:
error_msg = "Inkog scan timed out after 5 minutes"
logger.error(error_msg)
return {"error": error_msg}
except Exception as e:
error_msg = f"Inkog scan error: {str(e)}"
logger.error(error_msg)
return {"error": error_msg}
def skill_scan(
self,
path: str = ".",
output_format: str = "json",
deep: bool = False,
repo: Optional[str] = None
) -> Union[Dict[str, Any], str]:
"""Scan SKILL.md packages and agent tools for security vulnerabilities.
Args:
path: Path to scan or server name
output_format: Output format ("json", "text", "html", "sarif")
deep: Enable deep scan with AI orchestrator analysis
repo: Repository URL to scan
Returns:
Scan results as dict (JSON format) or string (other formats)
"""
is_available, error = _check_inkog_available(self._get_api_key())
if not is_available:
logger.error(error)
return {"error": error}
try:
# Build command
cmd = ["inkog", "skill-scan"]
if deep:
cmd.append("-deep")
cmd.extend(["-output", output_format])
if repo:
cmd.extend(["-repo", repo])
else:
cmd.append(path)
# Set environment
env = os.environ.copy()
api_key = self._get_api_key()
if api_key:
env["INKOG_API_KEY"] = api_key
# Execute scan
result = subprocess.run(
cmd,
capture_output=True,
text=True,
env=env,
timeout=300
)
if result.returncode != 0:
error_msg = f"Inkog skill scan failed: {result.stderr or result.stdout}"
logger.error(error_msg)
return {"error": error_msg}
# Parse output based on format
if output_format == "json":
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
return {"error": f"Failed to parse JSON output: {e}", "raw_output": result.stdout}
else:
return result.stdout
except subprocess.TimeoutExpired:
error_msg = "Inkog skill scan timed out after 5 minutes"
logger.error(error_msg)
return {"error": error_msg}
except Exception as e:
error_msg = f"Inkog skill scan error: {str(e)}"
logger.error(error_msg)
return {"error": error_msg}
def mcp_scan(
self,
server_name: str,
output_format: str = "json",
deep: bool = False,
repo: Optional[str] = None
) -> Union[Dict[str, Any], str]:
"""Scan MCP servers for security vulnerabilities.
Args:
server_name: MCP server name to scan
output_format: Output format ("json", "text", "html", "sarif")
deep: Enable deep scan with AI orchestrator analysis
repo: Repository URL for the MCP server source code
Returns:
Scan results as dict (JSON format) or string (other formats)
"""
is_available, error = _check_inkog_available(self._get_api_key())
if not is_available:
logger.error(error)
return {"error": error}
try:
# Build command
cmd = ["inkog", "mcp-scan"]
if deep:
cmd.append("-deep")
cmd.extend(["-output", output_format])
if repo:
cmd.extend(["-repo", repo])
cmd.append(server_name)
# Set environment
env = os.environ.copy()
api_key = self._get_api_key()
if api_key:
env["INKOG_API_KEY"] = api_key
# Execute scan
result = subprocess.run(
cmd,
capture_output=True,
text=True,
env=env,
timeout=300
)
if result.returncode != 0:
error_msg = f"Inkog MCP scan failed: {result.stderr or result.stdout}"
logger.error(error_msg)
return {"error": error_msg}
# Parse output based on format
if output_format == "json":
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
return {"error": f"Failed to parse JSON output: {e}", "raw_output": result.stdout}
else:
return result.stdout
except subprocess.TimeoutExpired:
error_msg = "Inkog MCP scan timed out after 5 minutes"
logger.error(error_msg)
return {"error": error_msg}
except Exception as e:
error_msg = f"Inkog MCP scan error: {str(e)}"
logger.error(error_msg)
return {"error": error_msg}
def analyze_findings(self, scan_results: Union[Dict[str, Any], str]) -> str:
"""Analyze scan results and provide a summary.
Args:
scan_results: Results from a JSON format scan. Non-dict inputs
(e.g. text/HTML output) are returned as-is.
Returns:
Human-readable analysis summary
"""
try:
if not isinstance(scan_results, dict):
return str(scan_results)
if "error" in scan_results:
return f"Scan failed: {scan_results['error']}"
summary = scan_results.get("summary", {})
total_findings = summary.get("total_findings", 0)
if total_findings == 0:
return "✅ No security vulnerabilities found in the agent code."
# Build severity breakdown
critical = summary.get("critical", 0)
high = summary.get("high", 0)
medium = summary.get("medium", 0)
low = summary.get("low", 0)
analysis = "🔍 Security Analysis Results:\n"
analysis += f"Total Findings: {total_findings}\n\n"
if critical > 0:
analysis += f"🔴 CRITICAL: {critical} findings - Immediate attention required\n"
if high > 0:
analysis += f"🟠 HIGH: {high} findings - Should be addressed soon\n"
if medium > 0:
analysis += f"🟡 MEDIUM: {medium} findings - Review recommended\n"
if low > 0:
analysis += f"⚪ LOW: {low} findings - Minor issues\n"
# Add key finding types if available
findings = scan_results.get("server_findings", [])
if findings:
patterns = set()
for finding in findings[:5]: # Show top 5 patterns
pattern = finding.get("pattern", "Unknown")
patterns.add(pattern)
if patterns:
analysis += "\nKey Vulnerability Types Detected:\n"
for pattern in sorted(patterns):
analysis += f"• {pattern}\n"
# Risk recommendations
if critical > 0 or high > 0:
analysis += "\n⚠️ Recommendation: Address critical and high severity findings before deployment."
return analysis
except Exception as e:
return f"Error analyzing scan results: {e}"
# Standalone functions for direct import and use
def scan_agent_code(
path: str = ".",
output_format: str = "json",
policy: str = "balanced",
severity: str = "low",
deep: bool = False
) -> Union[Dict[str, Any], str]:
"""Scan AI agent code for security vulnerabilities using Inkog.
Args:
path: Path to scan (directory or file)
output_format: Output format ("json", "text", "html", "sarif")
policy: Security policy ("low-noise", "balanced", "comprehensive")
severity: Minimum severity level ("critical", "high", "medium", "low")
deep: Enable deep scan with AI orchestrator analysis
Returns:
Scan results as dict (JSON format) or string (other formats)
"""
tool = InkogTool()
return tool.scan_directory(
path=path,
output_format=output_format,
policy=policy,
severity=severity,
deep=deep
)
def scan_skill_package(
path: str = ".",
deep: bool = False,
repo: Optional[str] = None
) -> Dict[str, Any]:
"""Scan SKILL.md packages for security vulnerabilities.
Args:
path: Path to scan
deep: Enable deep scan with AI orchestrator analysis
repo: Repository URL to scan
Returns:
Scan results as dict
"""
tool = InkogTool()
return tool.skill_scan(path=path, deep=deep, repo=repo)
def scan_mcp_server(
server_name: str,
deep: bool = False,
repo: Optional[str] = None
) -> Dict[str, Any]:
"""Scan MCP server for security vulnerabilities.
Args:
server_name: MCP server name to scan
deep: Enable deep scan with AI orchestrator analysis
repo: Repository URL for the MCP server
Returns:
Scan results as dict
"""
tool = InkogTool()
return tool.mcp_scan(server_name=server_name, deep=deep, repo=repo)
# Example usage and testing
if __name__ == "__main__":
print("\n" + "="*60)
print("Inkog Security Analysis Tool Demonstration")
print("="*60 + "\n")
# Check if inkog is available
is_available, error = _check_inkog_available()
if not is_available:
print(f"❌ Error: {error}")
print("\nTo use Inkog tools:")
print("1. Install CLI: brew tap inkog-io/inkog && brew install inkog")
print("2. Get API key: https://app.inkog.io")
print("3. Set environment: export INKOG_API_KEY=your_key_here")
else:
print("✅ Inkog is available and configured!")
# Example usage with current directory
try:
print("\nScanning current directory for security vulnerabilities...")
result = scan_agent_code(".", output_format="json", policy="balanced")
if isinstance(result, dict) and "error" not in result:
tool = InkogTool()
analysis = tool.analyze_findings(result)
print(analysis)
else:
print("Scan completed:", result)
except Exception as e:
print(f"Demo error: {e}")
@@ -0,0 +1,253 @@
"""NEXUS Prediction Market Tool for PraisonAI
Provides access to live Kalshi/Polymarket prediction market data and arbitrage opportunities
through the NEXUS API (https://nexus-agent-xa12.onrender.com).
Features:
- Free market data queries
- Paid arbitrage opportunity detection
- Real-time prediction market insights
API Documentation:
- Free endpoint: /kalshi?market={market_name}
- Paid endpoint: /arb/check?markets={market1,market2}
- Validation: https://agentic.market/validate?url=https://nexus-agent-xa12.onrender.com/arb/check
"""
from typing import Optional, List, Dict, Any
import requests
import json
from praisonai_tools.tools.base import BaseTool
class NexusPredictionMarketTool(BaseTool):
"""Tool for accessing NEXUS prediction market data and arbitrage opportunities."""
name: str = "nexus_prediction_market"
description: str = """Access live Kalshi/Polymarket prediction market data and arbitrage opportunities.
Features:
- Query free market data for specific markets (Fed, BTC, etc.)
- Check arbitrage opportunities between prediction markets (requires payment)
- Get real-time market insights and pricing data
Use cases:
- Market research and analysis
- Arbitrage opportunity identification
- Prediction market trend monitoring
"""
def __init__(self, source_name: Optional[str] = None, api_key: Optional[str] = None):
"""Initialize the NEXUS Prediction Market Tool.
Args:
source_name: Optional source identifier for tracking (sent as X-NEXUS-Source header)
api_key: Optional API key for paid features (if implemented)
"""
super().__init__()
self.base_url = "https://nexus-agent-xa12.onrender.com"
self.source_name = source_name or "praisonai-agent"
self.api_key = api_key
self.headers = {
"X-NEXUS-Source": self.source_name,
"User-Agent": "PraisonAI-NEXUS-Tool/1.0",
"Accept": "application/json"
}
if self.api_key:
self.headers["Authorization"] = f"Bearer {self.api_key}"
def get_market_data(self, market: str) -> Dict[str, Any]:
"""Get free market data for a specific prediction market.
Args:
market: Market identifier (e.g., 'Fed', 'BTC', 'Election')
Returns:
Dict containing market data including prices, volumes, and trends
"""
try:
url = f"{self.base_url}/kalshi"
params = {"market": market}
response = requests.get(url, params=params, headers=self.headers, timeout=30)
response.raise_for_status()
return {
"success": True,
"market": market,
"data": response.json(),
"source": "nexus-free-api"
}
except requests.RequestException as e:
return {
"success": False,
"error": f"Failed to fetch market data: {str(e)}",
"market": market
}
except json.JSONDecodeError as e:
return {
"success": False,
"error": f"Invalid JSON response: {str(e)}",
"market": market
}
def check_arbitrage_opportunities(self, markets: List[str]) -> Dict[str, Any]:
"""Check for arbitrage opportunities between prediction markets (paid feature).
Args:
markets: List of market identifiers to analyze for arbitrage
Returns:
Dict containing arbitrage opportunities and analysis
Note:
This is a paid feature requiring $0.02 USDC on eip155:8453
"""
try:
url = f"{self.base_url}/arb/check"
markets_param = ",".join(markets)
params = {"markets": markets_param}
response = requests.get(url, params=params, headers=self.headers, timeout=30)
response.raise_for_status()
return {
"success": True,
"markets": markets,
"arbitrage_data": response.json(),
"source": "nexus-paid-api",
"note": "This is a paid feature ($0.02 USDC on eip155:8453)"
}
except requests.RequestException as e:
if e.response and e.response.status_code == 402:
return {
"success": False,
"error": "Payment required for arbitrage analysis. Cost: $0.02 USDC on eip155:8453",
"markets": markets,
"payment_info": "This feature requires payment on eip155:8453 blockchain"
}
return {
"success": False,
"error": f"Failed to check arbitrage opportunities: {str(e)}",
"markets": markets
}
except json.JSONDecodeError as e:
return {
"success": False,
"error": f"Invalid JSON response: {str(e)}",
"markets": markets
}
def get_agent_info(self) -> Dict[str, Any]:
"""Get NEXUS agent information and capabilities.
Returns:
Dict containing agent metadata and API information
"""
try:
url = f"{self.base_url}/.well-known/agent.json"
response = requests.get(url, headers=self.headers, timeout=30)
response.raise_for_status()
return {
"success": True,
"agent_info": response.json(),
"source": "nexus-agent-metadata"
}
except requests.RequestException as e:
return {
"success": False,
"error": f"Failed to fetch agent info: {str(e)}"
}
except json.JSONDecodeError as e:
return {
"success": False,
"error": f"Invalid JSON response: {str(e)}"
}
def run(self,
action: str,
market: Optional[str] = None,
markets: Optional[List[str]] = None) -> str:
"""Execute NEXUS prediction market operations.
Args:
action: Action to perform ('get_market_data', 'check_arbitrage', 'get_agent_info')
market: Single market identifier (for get_market_data)
markets: List of markets (for check_arbitrage)
Returns:
JSON string with operation results
"""
try:
if action == "get_market_data":
if not market:
return json.dumps({
"success": False,
"error": "Market parameter required for get_market_data action"
})
result = self.get_market_data(market)
elif action == "check_arbitrage":
if not markets or not isinstance(markets, list):
return json.dumps({
"success": False,
"error": "Markets parameter (list) required for check_arbitrage action"
})
result = self.check_arbitrage_opportunities(markets)
elif action == "get_agent_info":
result = self.get_agent_info()
else:
return json.dumps({
"success": False,
"error": f"Unknown action: {action}. Available actions: get_market_data, check_arbitrage, get_agent_info"
})
return json.dumps(result, indent=2)
except Exception as e:
return json.dumps({
"success": False,
"error": f"Unexpected error: {str(e)}"
})
# For backward compatibility and ease of use
def nexus_prediction_market_tool(
action: str,
market: Optional[str] = None,
markets: Optional[List[str]] = None,
source_name: Optional[str] = None
) -> str:
"""Function-based interface for NEXUS prediction market operations.
Args:
action: Action to perform ('get_market_data', 'check_arbitrage', 'get_agent_info')
market: Single market identifier (for get_market_data)
markets: List of markets (for check_arbitrage)
source_name: Optional source identifier for tracking
Returns:
JSON string with operation results
Examples:
# Get Fed market data
result = nexus_prediction_market_tool("get_market_data", market="Fed")
# Check arbitrage between Fed and BTC markets
result = nexus_prediction_market_tool("check_arbitrage", markets=["Fed", "BTC"])
# Get agent info
result = nexus_prediction_market_tool("get_agent_info")
"""
tool = NexusPredictionMarketTool(source_name=source_name)
return tool.run(action=action, market=market, markets=markets)
__all__ = ["NexusPredictionMarketTool", "nexus_prediction_market_tool"]
+259
View File
@@ -0,0 +1,259 @@
"""SwarmScore integration tool for PraisonAI.
SwarmScore is a portable trust rating system for AI agents based on verified execution history.
This tool allows agents to:
- Load their SwarmScore rating and certificate
- Verify SwarmScore freshness
- Access discovery manifest for agent-to-agent lookup
More information: https://swarmsync.ai/docs/protocol-specs/swarmscore
"""
import json
import logging
from typing import Dict, Any
from urllib.parse import urlparse, quote
try:
import requests
except ImportError:
requests = None
from .base import BaseTool, ToolResult
logger = logging.getLogger(__name__)
class SwarmScoreTool(BaseTool):
"""Tool for integrating with SwarmScore trust ratings."""
name = "swarmscore"
description = "Load and verify SwarmScore trust ratings for AI agents"
def __init__(self, api_base_url: str = "https://api.swarmsync.ai/v1/swarmscore/"):
"""Initialize SwarmScore tool.
Args:
api_base_url: Base URL for SwarmScore API
"""
if requests is None:
raise ImportError("requests is required for SwarmScoreTool. Install with: pip install requests")
super().__init__()
self.api_base_url = api_base_url.rstrip('/')
def load_swarmscore(self, slug: str) -> ToolResult:
"""Load SwarmScore data by agent slug.
Args:
slug: Agent slug identifier
Returns:
ToolResult containing SwarmScore data including passport, certificate, and verification payload
"""
try:
url = f"{self.api_base_url}/load-by-slug/{quote(slug, safe='')}"
response = requests.get(url, timeout=30)
response.raise_for_status()
data = response.json()
return ToolResult(
output=data,
success=True,
metadata={"message": f"Successfully loaded SwarmScore for agent '{slug}'"}
)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON response for SwarmScore {slug}: {e}")
return ToolResult(
output=None,
success=False,
error="Invalid response format from SwarmScore API"
)
except requests.exceptions.RequestException as e:
logger.error(f"Failed to load SwarmScore for {slug}: {e}")
return ToolResult(
output=None,
success=False,
error=f"Failed to load SwarmScore: {str(e)}"
)
def verify_swarmscore(self, verify_payload: Dict[str, Any]) -> ToolResult:
"""Verify SwarmScore freshness using verification payload.
Args:
verify_payload: Verification payload from load_swarmscore response
Returns:
ToolResult containing verification status
"""
try:
url = f"{self.api_base_url}/verify"
response = requests.post(
url,
json=verify_payload,
headers={"Content-Type": "application/json"},
timeout=30
)
response.raise_for_status()
data = response.json()
return ToolResult(
output=data,
success=True,
metadata={"message": "SwarmScore verification completed"}
)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON response for SwarmScore verification: {e}")
return ToolResult(
output=None,
success=False,
error="Invalid response format from verification API"
)
except requests.exceptions.RequestException as e:
logger.error(f"Failed to verify SwarmScore: {e}")
return ToolResult(
output=None,
success=False,
error=f"Failed to verify SwarmScore: {str(e)}"
)
def get_discovery_manifest(self) -> ToolResult:
"""Get machine-readable agent discovery manifest.
Returns:
ToolResult containing discovery manifest data
"""
try:
parsed = urlparse(self.api_base_url)
base = f"{parsed.scheme}://{parsed.netloc}"
url = f"{base}/.well-known/agent-card.json"
response = requests.get(url, timeout=30)
response.raise_for_status()
data = response.json()
return ToolResult(
output=data,
success=True,
metadata={"message": "Retrieved SwarmScore discovery manifest"}
)
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON in discovery manifest: {e}")
return ToolResult(
output=None,
success=False,
error="Invalid discovery manifest format"
)
except requests.exceptions.RequestException as e:
logger.error(f"Failed to get discovery manifest: {e}")
return ToolResult(
output=None,
success=False,
error=f"Failed to get discovery manifest: {str(e)}"
)
def run(self, action: str = "load", **kwargs) -> ToolResult:
"""Execute SwarmScore action.
Args:
action: Action to perform ('load', 'verify', or 'discover')
**kwargs: Action-specific parameters
Returns:
ToolResult with action results
"""
if action == "load":
slug = kwargs.get("slug")
if not slug:
return ToolResult(
output=None,
success=False,
error="Missing required parameter 'slug' for load action"
)
return self.load_swarmscore(slug)
elif action == "verify":
verify_payload = kwargs.get("verify_payload")
if not verify_payload:
return ToolResult(
output=None,
success=False,
error="Missing required parameter 'verify_payload' for verify action"
)
return self.verify_swarmscore(verify_payload)
elif action == "discover":
return self.get_discovery_manifest()
else:
return ToolResult(
output=None,
success=False,
error=f"Unknown action '{action}'. Supported actions: load, verify, discover"
)
# Standalone functions for easy agent integration
def load_swarmscore_by_slug(slug: str) -> Dict[str, Any]:
"""Load SwarmScore data by agent slug.
Args:
slug: Agent slug identifier
Returns:
Dictionary containing SwarmScore data
Raises:
Exception: If loading fails
"""
tool = SwarmScoreTool()
result = tool.load_swarmscore(slug)
if not result.success:
raise Exception(result.error)
return result.output
def verify_swarmscore_freshness(verify_payload: Dict[str, Any]) -> Dict[str, Any]:
"""Verify SwarmScore freshness.
Args:
verify_payload: Verification payload from load response
Returns:
Dictionary containing verification results
Raises:
Exception: If verification fails
"""
tool = SwarmScoreTool()
result = tool.verify_swarmscore(verify_payload)
if not result.success:
raise Exception(result.error)
return result.output
def get_agent_discovery_manifest() -> Dict[str, Any]:
"""Get machine-readable agent discovery manifest.
Returns:
Dictionary containing discovery manifest
Raises:
Exception: If retrieval fails
"""
tool = SwarmScoreTool()
result = tool.get_discovery_manifest()
if not result.success:
raise Exception(result.error)
return result.output
+4 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-tools"
version = "0.2.30"
version = "0.2.32"
description = "Extended tools for PraisonAI Agents"
authors = [
{name = "Mervin Praison"}
@@ -57,6 +57,9 @@ n8n = [
langextract = [
"langextract>=0.1.0",
]
swarmscore = [
"requests>=2.28.0",
]
video-motion = [
"playwright>=1.40",
"imageio-ffmpeg>=0.5",
+305
View File
@@ -0,0 +1,305 @@
"""Tests for Inkog Security Analysis Tool."""
from unittest.mock import patch, MagicMock
from praisonai_tools.tools.inkog_tool import InkogTool, _check_inkog_available, scan_agent_code
class TestInkogTool:
"""Test suite for InkogTool."""
def test_init(self):
"""Test InkogTool initialization."""
# Test default initialization
tool = InkogTool()
assert tool.name == "inkog_security"
assert "security" in tool.description.lower()
# Test with custom parameters
tool = InkogTool(api_key="test_key", server_url="http://test.com")
assert tool._api_key == "test_key"
assert tool._server_url == "http://test.com"
@patch.dict("os.environ", {"INKOG_API_KEY": "test_env_key"})
def test_get_api_key(self):
"""Test API key retrieval."""
tool = InkogTool()
assert tool._get_api_key() == "test_env_key"
# Instance key takes precedence
tool = InkogTool(api_key="instance_key")
assert tool._get_api_key() == "instance_key"
@patch.dict("os.environ", {"INKOG_SERVER_URL": "http://env.com"})
def test_get_server_url(self):
"""Test server URL retrieval."""
tool = InkogTool()
assert tool._get_server_url() == "http://env.com"
# Instance URL takes precedence
tool = InkogTool(server_url="http://instance.com")
assert tool._get_server_url() == "http://instance.com"
def test_analyze_findings_empty(self):
"""Test analyze_findings with empty results."""
tool = InkogTool()
# Test with error
error_results = {"error": "test error"}
analysis = tool.analyze_findings(error_results)
assert "Scan failed" in analysis
assert "test error" in analysis
# Test with no findings
empty_results = {"summary": {"total_findings": 0}}
analysis = tool.analyze_findings(empty_results)
assert "No security vulnerabilities found" in analysis
def test_analyze_findings_with_data(self):
"""Test analyze_findings with security findings."""
tool = InkogTool()
results = {
"summary": {
"total_findings": 5,
"critical": 2,
"high": 2,
"medium": 1,
"low": 0
},
"server_findings": [
{"pattern": "Token Bombing"},
{"pattern": "Prompt Injection"},
{"pattern": "Recursive Tool Calling"}
]
}
analysis = tool.analyze_findings(results)
assert "Total Findings: 5" in analysis
assert "CRITICAL: 2 findings" in analysis
assert "HIGH: 2 findings" in analysis
assert "MEDIUM: 1 findings" in analysis
assert "Token Bombing" in analysis
assert "Immediate attention required" in analysis
def test_analyze_findings_string_input(self):
"""String (text/HTML) scan output is returned as-is, even when it
contains the word 'error' (regression: no TypeError)."""
tool = InkogTool()
text_output = "Finding: potential error in tool loop termination."
analysis = tool.analyze_findings(text_output)
assert analysis == text_output
class TestInkogAvailability:
"""Test inkog availability checking."""
@patch("subprocess.run")
def test_check_inkog_available_success(self, mock_run):
"""Test successful inkog availability check."""
# Mock successful version check
mock_run.return_value = MagicMock(returncode=0)
with patch.dict("os.environ", {"INKOG_API_KEY": "test_key"}):
is_available, error = _check_inkog_available()
assert is_available is True
assert error is None
@patch("subprocess.run")
def test_check_inkog_not_installed(self, mock_run):
"""Test inkog not installed."""
mock_run.side_effect = FileNotFoundError()
is_available, error = _check_inkog_available()
assert is_available is False
assert "not installed" in error.lower()
@patch("subprocess.run")
def test_check_inkog_no_api_key(self, mock_run):
"""Test inkog available but no API key."""
mock_run.return_value = MagicMock(returncode=0)
with patch.dict("os.environ", {}, clear=True):
is_available, error = _check_inkog_available()
assert is_available is False
assert "INKOG_API_KEY" in error
@patch("subprocess.run")
def test_check_inkog_available_with_instance_api_key(self, mock_run):
"""Instance-provided api_key satisfies the availability check."""
mock_run.return_value = MagicMock(returncode=0)
with patch.dict("os.environ", {}, clear=True):
is_available, error = _check_inkog_available(api_key="instance_key")
assert is_available is True
assert error is None
class TestInkogScanOperations:
"""Test inkog scan operations."""
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
@patch("subprocess.run")
@patch.dict("os.environ", {"INKOG_API_KEY": "test_key"})
def test_scan_directory_json_success(self, mock_run, mock_check):
"""Test successful directory scan with JSON output."""
mock_check.return_value = (True, None)
mock_run.return_value = MagicMock(
returncode=0,
stdout='{"summary": {"total_findings": 0}}'
)
tool = InkogTool()
result = tool.scan_directory(path="./test", output_format="json")
assert isinstance(result, dict)
assert "summary" in result
assert result["summary"]["total_findings"] == 0
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
@patch("subprocess.run")
@patch.dict("os.environ", {"INKOG_API_KEY": "test_key"})
def test_scan_directory_text_success(self, mock_run, mock_check):
"""Test successful directory scan with text output."""
mock_check.return_value = (True, None)
mock_run.return_value = MagicMock(
returncode=0,
stdout="No vulnerabilities found."
)
tool = InkogTool()
result = tool.scan_directory(path="./test", output_format="text")
assert isinstance(result, str)
assert "vulnerabilities" in result.lower()
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
def test_scan_directory_not_available(self, mock_check):
"""Test scan when inkog is not available."""
mock_check.return_value = (False, "inkog not installed")
tool = InkogTool()
result = tool.scan_directory()
assert isinstance(result, dict)
assert "error" in result
assert "not installed" in result["error"]
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
@patch("subprocess.run")
def test_scan_directory_command_failure(self, mock_run, mock_check):
"""Test scan when inkog command fails."""
mock_check.return_value = (True, None)
mock_run.return_value = MagicMock(
returncode=1,
stderr="Command failed",
stdout=""
)
tool = InkogTool()
result = tool.scan_directory()
assert isinstance(result, dict)
assert "error" in result
assert "failed" in result["error"].lower()
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
@patch("subprocess.run")
@patch.dict("os.environ", {"INKOG_API_KEY": "test_key"})
def test_skill_scan_success(self, mock_run, mock_check):
"""Test successful skill package scan."""
mock_check.return_value = (True, None)
mock_run.return_value = MagicMock(
returncode=0,
stdout='{"scan_type": "skill", "findings": []}'
)
tool = InkogTool()
result = tool.skill_scan(path="./skill", deep=True)
assert isinstance(result, dict)
assert "scan_type" in result or "findings" in result
@patch("praisonai_tools.tools.inkog_tool._check_inkog_available")
@patch("subprocess.run")
@patch.dict("os.environ", {"INKOG_API_KEY": "test_key"})
def test_mcp_scan_success(self, mock_run, mock_check):
"""Test successful MCP server scan."""
mock_check.return_value = (True, None)
mock_run.return_value = MagicMock(
returncode=0,
stdout='{"mcp_server": "github", "vulnerabilities": []}'
)
tool = InkogTool()
result = tool.mcp_scan("github", deep=True)
assert isinstance(result, dict)
# Should have some scan results structure
def test_run_method_routing(self):
"""Test the run method routes to correct functions."""
tool = InkogTool()
with patch.object(tool, 'scan_directory') as mock_scan:
mock_scan.return_value = {"test": "result"}
tool.run(action="scan", path="./test")
mock_scan.assert_called_once_with(
path="./test",
output_format="json",
policy="balanced",
severity="low",
deep=False
)
with patch.object(tool, 'skill_scan') as mock_skill:
mock_skill.return_value = {"test": "result"}
tool.run(action="skill-scan", path="./test")
mock_skill.assert_called_once_with(
path="./test",
output_format="json",
deep=False,
repo=None
)
class TestStandaloneFunctions:
"""Test standalone function interfaces."""
@patch("praisonai_tools.tools.inkog_tool.InkogTool")
def test_scan_agent_code(self, mock_tool_class):
"""Test scan_agent_code standalone function."""
mock_tool = MagicMock()
mock_tool.scan_directory.return_value = {"result": "test"}
mock_tool_class.return_value = mock_tool
scan_agent_code("./test", deep=True)
mock_tool_class.assert_called_once()
mock_tool.scan_directory.assert_called_once_with(
path="./test",
output_format="json",
policy="balanced",
severity="low",
deep=True
)
if __name__ == "__main__":
# Run basic smoke test
print("Running Inkog Tool smoke tests...")
# Test basic initialization
tool = InkogTool()
print(f"✓ Tool name: {tool.name}")
print(f"✓ Tool description contains 'security': {'security' in tool.description.lower()}")
# Test availability check (will likely fail without actual installation)
is_available, error = _check_inkog_available()
if is_available:
print("✓ Inkog is available and configured")
else:
print(f"⚠ Inkog not available: {error}")
print("This is expected if inkog CLI is not installed")
print("Smoke tests completed!")
+244
View File
@@ -0,0 +1,244 @@
"""Tests for NEXUS Prediction Market Tool"""
import pytest
import json
import requests
from unittest.mock import Mock, patch
from praisonai_tools.tools.nexus_prediction_market_tool import (
NexusPredictionMarketTool,
nexus_prediction_market_tool
)
class TestNexusPredictionMarketTool:
"""Test cases for NEXUS Prediction Market Tool"""
def test_tool_initialization(self):
"""Test tool initialization with default and custom parameters."""
# Default initialization
tool = NexusPredictionMarketTool()
assert tool.name == "nexus_prediction_market"
assert tool.source_name == "praisonai-agent"
assert "X-NEXUS-Source" in tool.headers
assert tool.base_url == "https://nexus-agent-xa12.onrender.com"
# Custom initialization
custom_tool = NexusPredictionMarketTool(
source_name="test-bot",
api_key="test-key"
)
assert custom_tool.source_name == "test-bot"
assert custom_tool.headers["X-NEXUS-Source"] == "test-bot"
assert "Authorization" in custom_tool.headers
@patch('praisonai_tools.tools.nexus_prediction_market_tool.requests.get')
def test_get_market_data_success(self, mock_get):
"""Test successful market data retrieval."""
# Mock successful response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"market": "Fed",
"price": 0.75,
"volume": 1000000,
"trend": "bullish"
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
tool = NexusPredictionMarketTool()
result = tool.get_market_data("Fed")
assert result["success"] is True
assert result["market"] == "Fed"
assert "data" in result
assert result["source"] == "nexus-free-api"
# Verify request was made correctly
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
assert "kalshi" in args[0]
assert kwargs["params"]["market"] == "Fed"
assert "X-NEXUS-Source" in kwargs["headers"]
@patch('praisonai_tools.tools.nexus_prediction_market_tool.requests.get')
def test_get_market_data_failure(self, mock_get):
"""Test market data retrieval failure."""
# Mock failed response
mock_get.side_effect = requests.exceptions.ConnectionError("Network error")
tool = NexusPredictionMarketTool()
result = tool.get_market_data("Fed")
assert result["success"] is False
assert "error" in result
assert result["market"] == "Fed"
@patch('praisonai_tools.tools.nexus_prediction_market_tool.requests.get')
def test_check_arbitrage_opportunities_success(self, mock_get):
"""Test successful arbitrage checking."""
# Mock successful response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"opportunities": [
{
"markets": ["Fed", "BTC"],
"profit_potential": 0.05,
"risk_level": "medium"
}
]
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
tool = NexusPredictionMarketTool()
result = tool.check_arbitrage_opportunities(["Fed", "BTC"])
assert result["success"] is True
assert result["markets"] == ["Fed", "BTC"]
assert "arbitrage_data" in result
assert result["source"] == "nexus-paid-api"
# Verify request was made correctly
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
assert "arb/check" in args[0]
assert kwargs["params"]["markets"] == "Fed,BTC"
@patch('praisonai_tools.tools.nexus_prediction_market_tool.requests.get')
def test_check_arbitrage_payment_required(self, mock_get):
"""Test arbitrage checking when payment is required."""
# Mock payment required response
mock_response = Mock()
mock_response.status_code = 402
http_error = requests.exceptions.HTTPError("Payment Required")
http_error.response = mock_response
mock_get.side_effect = http_error
tool = NexusPredictionMarketTool()
result = tool.check_arbitrage_opportunities(["Fed", "BTC"])
assert result["success"] is False
assert "payment required" in result["error"].lower()
assert result["markets"] == ["Fed", "BTC"]
@patch('praisonai_tools.tools.nexus_prediction_market_tool.requests.get')
def test_get_agent_info_success(self, mock_get):
"""Test successful agent info retrieval."""
# Mock successful response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"name": "NEXUS",
"description": "Prediction market data API",
"version": "1.0"
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
tool = NexusPredictionMarketTool()
result = tool.get_agent_info()
assert result["success"] is True
assert "agent_info" in result
assert result["source"] == "nexus-agent-metadata"
# Verify request was made correctly
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
assert ".well-known/agent.json" in args[0]
def test_run_method_get_market_data(self):
"""Test run method with get_market_data action."""
tool = NexusPredictionMarketTool()
with patch.object(tool, 'get_market_data') as mock_method:
mock_method.return_value = {"success": True, "test": "data"}
result = tool.run("get_market_data", market="Fed")
result_data = json.loads(result)
assert result_data["success"] is True
mock_method.assert_called_once_with("Fed")
def test_run_method_check_arbitrage(self):
"""Test run method with check_arbitrage action."""
tool = NexusPredictionMarketTool()
with patch.object(tool, 'check_arbitrage_opportunities') as mock_method:
mock_method.return_value = {"success": True, "test": "data"}
result = tool.run("check_arbitrage", markets=["Fed", "BTC"])
result_data = json.loads(result)
assert result_data["success"] is True
mock_method.assert_called_once_with(["Fed", "BTC"])
def test_run_method_get_agent_info(self):
"""Test run method with get_agent_info action."""
tool = NexusPredictionMarketTool()
with patch.object(tool, 'get_agent_info') as mock_method:
mock_method.return_value = {"success": True, "test": "data"}
result = tool.run("get_agent_info")
result_data = json.loads(result)
assert result_data["success"] is True
mock_method.assert_called_once()
def test_run_method_invalid_action(self):
"""Test run method with invalid action."""
tool = NexusPredictionMarketTool()
result = tool.run("invalid_action")
result_data = json.loads(result)
assert result_data["success"] is False
assert "unknown action" in result_data["error"].lower()
def test_run_method_missing_parameters(self):
"""Test run method with missing required parameters."""
tool = NexusPredictionMarketTool()
# Missing market parameter
result = tool.run("get_market_data")
result_data = json.loads(result)
assert result_data["success"] is False
assert "market parameter required" in result_data["error"].lower()
# Missing markets parameter
result = tool.run("check_arbitrage")
result_data = json.loads(result)
assert result_data["success"] is False
assert "markets parameter" in result_data["error"].lower()
class TestNexusPredictionMarketFunction:
"""Test cases for the function-based interface"""
def test_function_interface(self):
"""Test the function-based interface."""
with patch('praisonai_tools.tools.nexus_prediction_market_tool.NexusPredictionMarketTool') as mock_class:
mock_instance = Mock()
mock_instance.run.return_value = '{"success": true}'
mock_class.return_value = mock_instance
result = nexus_prediction_market_tool(
action="get_market_data",
market="Fed",
source_name="test-source"
)
assert result == '{"success": true}'
mock_class.assert_called_once_with(source_name="test-source")
mock_instance.run.assert_called_once_with(
action="get_market_data",
market="Fed",
markets=None
)
if __name__ == "__main__":
pytest.main([__file__])
+451
View File
@@ -0,0 +1,451 @@
"""Tests for SwarmScore tool integration."""
import json
import pytest
import requests as real_requests
from urllib.parse import urlparse
from unittest.mock import Mock, patch
from praisonai_tools.tools.swarmscore_tool import (
SwarmScoreTool,
load_swarmscore_by_slug,
verify_swarmscore_freshness,
get_agent_discovery_manifest
)
from praisonai_tools.tools.base import ToolResult
class TestSwarmScoreTool:
"""Test cases for SwarmScoreTool class."""
def setup_method(self):
"""Set up test fixtures."""
self.tool = SwarmScoreTool()
self.sample_score_data = {
"score": 85,
"tier": "gold",
"jobs_completed": 150,
"success_rate": 94.5,
"verify_payload": {
"signature": "test_signature",
"timestamp": "2024-01-15T10:30:00Z",
"agent_id": "test-agent-123"
}
}
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_load_swarmscore_success(self, mock_requests):
"""Test successful SwarmScore loading."""
# Setup mock response
mock_response = Mock()
mock_response.json.return_value = self.sample_score_data
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
# Execute
result = self.tool.load_swarmscore("test-agent")
# Verify
assert result.success is True
assert result.output == self.sample_score_data
assert "test-agent" in result.metadata["message"]
mock_requests.get.assert_called_once()
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_load_swarmscore_slug_url_encoded(self, mock_requests):
"""Slugs with special characters must be URL-encoded to avoid malformed paths."""
mock_response = Mock()
mock_response.json.return_value = self.sample_score_data
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
result = self.tool.load_swarmscore("acme/agent one")
assert result.success is True
called_url = mock_requests.get.call_args[0][0]
assert called_url.endswith("/load-by-slug/acme%2Fagent%20one")
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_load_swarmscore_network_error(self, mock_requests):
"""Test SwarmScore loading with network error."""
# Setup mock to raise exception
mock_requests.exceptions = real_requests.exceptions
mock_requests.get.side_effect = real_requests.exceptions.RequestException("Network error")
# Execute
result = self.tool.load_swarmscore("test-agent")
# Verify
assert result.success is False
assert "Network error" in result.error
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_load_swarmscore_invalid_json(self, mock_requests):
"""Test SwarmScore loading with invalid JSON response."""
# Setup mock response with invalid JSON
mock_response = Mock()
mock_requests.exceptions = real_requests.exceptions
mock_response.json.side_effect = json.JSONDecodeError("Invalid JSON", "", 0)
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
# Execute
result = self.tool.load_swarmscore("test-agent")
# Verify
assert result.success is False
assert "Invalid response format" in result.error
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_verify_swarmscore_success(self, mock_requests):
"""Test successful SwarmScore verification."""
# Setup mock response
verify_data = {"verified": True, "timestamp": "2024-01-15T10:30:00Z"}
mock_response = Mock()
mock_response.json.return_value = verify_data
mock_response.raise_for_status.return_value = None
mock_requests.post.return_value = mock_response
# Execute
verify_payload = self.sample_score_data["verify_payload"]
result = self.tool.verify_swarmscore(verify_payload)
# Verify
assert result.success is True
assert result.output == verify_data
mock_requests.post.assert_called_once()
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_verify_swarmscore_error(self, mock_requests):
"""Test SwarmScore verification with error."""
# Setup mock to raise exception
mock_requests.exceptions = real_requests.exceptions
mock_requests.post.side_effect = real_requests.exceptions.RequestException("Verification failed")
# Execute
result = self.tool.verify_swarmscore({"test": "payload"})
# Verify
assert result.success is False
assert "Verification failed" in result.error
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_get_discovery_manifest_success(self, mock_requests):
"""Test successful discovery manifest retrieval."""
# Setup mock response
manifest_data = {
"agents": ["agent1", "agent2"],
"capabilities": ["trading", "analysis"],
"version": "1.0"
}
mock_response = Mock()
mock_response.json.return_value = manifest_data
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
# Execute
result = self.tool.get_discovery_manifest()
# Verify
assert result.success is True
assert result.output == manifest_data
mock_requests.get.assert_called_once()
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_get_discovery_manifest_url_derivation(self, mock_requests):
"""Discovery manifest URL must resolve to the domain root regardless of version prefix."""
mock_response = Mock()
mock_response.json.return_value = {"agents": []}
mock_response.raise_for_status.return_value = None
mock_requests.get.return_value = mock_response
for base_url in [
"https://api.swarmsync.ai/v1/swarmscore/",
"https://api.swarmsync.ai/v2/swarmscore/",
"https://staging.swarmsync.ai/api/swarmscore/",
"https://api.swarmsync.ai/swarmscore",
]:
mock_requests.get.reset_mock()
tool = SwarmScoreTool(api_base_url=base_url)
result = tool.get_discovery_manifest()
assert result.success is True
called_url = mock_requests.get.call_args[0][0]
expected_root = f"{urlparse(base_url).scheme}://{urlparse(base_url).netloc}"
assert called_url == f"{expected_root}/.well-known/agent-card.json"
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_get_discovery_manifest_error(self, mock_requests):
"""Test discovery manifest retrieval with error."""
# Setup mock to raise exception
mock_requests.exceptions = real_requests.exceptions
mock_requests.get.side_effect = real_requests.exceptions.RequestException("Manifest error")
# Execute
result = self.tool.get_discovery_manifest()
# Verify
assert result.success is False
assert "Manifest error" in result.error
def test_run_load_action(self):
"""Test run method with load action."""
with patch.object(self.tool, 'load_swarmscore') as mock_load:
mock_load.return_value = ToolResult(output=self.sample_score_data, success=True)
result = self.tool.run("load", slug="test-agent")
assert result.success is True
mock_load.assert_called_once_with("test-agent")
def test_run_verify_action(self):
"""Test run method with verify action."""
with patch.object(self.tool, 'verify_swarmscore') as mock_verify:
mock_verify.return_value = ToolResult(output={"verified": True}, success=True)
verify_payload = {"test": "payload"}
result = self.tool.run("verify", verify_payload=verify_payload)
assert result.success is True
mock_verify.assert_called_once_with(verify_payload)
def test_run_discover_action(self):
"""Test run method with discover action."""
with patch.object(self.tool, 'get_discovery_manifest') as mock_discover:
mock_discover.return_value = ToolResult(output={"agents": []}, success=True)
result = self.tool.run("discover")
assert result.success is True
mock_discover.assert_called_once()
def test_run_default_action(self):
"""Bare run() defaults to 'load' and degrades gracefully without a slug."""
result = self.tool.run()
assert result.success is False
assert "Missing required parameter 'slug'" in result.error
def test_run_invalid_action(self):
"""Test run method with invalid action."""
result = self.tool.run("invalid_action")
assert result.success is False
assert "Unknown action" in result.error
def test_run_missing_parameters(self):
"""Test run method with missing required parameters."""
# Test load without slug
result = self.tool.run("load")
assert result.success is False
assert "Missing required parameter 'slug'" in result.error
# Test verify without verify_payload
result = self.tool.run("verify")
assert result.success is False
assert "Missing required parameter 'verify_payload'" in result.error
class TestStandaloneFunctions:
"""Test cases for standalone utility functions."""
def setup_method(self):
"""Set up test fixtures."""
self.sample_data = {
"score": 75,
"tier": "silver",
"verify_payload": {"test": "payload"}
}
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_load_swarmscore_by_slug_success(self, mock_tool_class):
"""Test successful standalone load function."""
# Setup mock
mock_tool = Mock()
mock_tool.load_swarmscore.return_value = ToolResult(
output=self.sample_data,
success=True
)
mock_tool_class.return_value = mock_tool
# Execute
result = load_swarmscore_by_slug("test-agent")
# Verify
assert result == self.sample_data
mock_tool.load_swarmscore.assert_called_once_with("test-agent")
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_load_swarmscore_by_slug_failure(self, mock_tool_class):
"""Test standalone load function with failure."""
# Setup mock
mock_tool = Mock()
mock_tool.load_swarmscore.return_value = ToolResult(
output=None,
success=False,
error="Load failed"
)
mock_tool_class.return_value = mock_tool
# Execute and verify exception
with pytest.raises(Exception, match="Load failed"):
load_swarmscore_by_slug("test-agent")
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_verify_swarmscore_freshness_success(self, mock_tool_class):
"""Test successful standalone verify function."""
# Setup mock
verify_data = {"verified": True}
mock_tool = Mock()
mock_tool.verify_swarmscore.return_value = ToolResult(
output=verify_data,
success=True
)
mock_tool_class.return_value = mock_tool
# Execute
payload = {"test": "payload"}
result = verify_swarmscore_freshness(payload)
# Verify
assert result == verify_data
mock_tool.verify_swarmscore.assert_called_once_with(payload)
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_verify_swarmscore_freshness_failure(self, mock_tool_class):
"""Test standalone verify function with failure."""
# Setup mock
mock_tool = Mock()
mock_tool.verify_swarmscore.return_value = ToolResult(
output=None,
success=False,
error="Verification failed"
)
mock_tool_class.return_value = mock_tool
# Execute and verify exception
with pytest.raises(Exception, match="Verification failed"):
verify_swarmscore_freshness({"test": "payload"})
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_get_agent_discovery_manifest_success(self, mock_tool_class):
"""Test successful standalone discovery function."""
# Setup mock
manifest_data = {"agents": ["test1", "test2"]}
mock_tool = Mock()
mock_tool.get_discovery_manifest.return_value = ToolResult(
output=manifest_data,
success=True
)
mock_tool_class.return_value = mock_tool
# Execute
result = get_agent_discovery_manifest()
# Verify
assert result == manifest_data
mock_tool.get_discovery_manifest.assert_called_once()
@patch('praisonai_tools.tools.swarmscore_tool.SwarmScoreTool')
def test_get_agent_discovery_manifest_failure(self, mock_tool_class):
"""Test standalone discovery function with failure."""
# Setup mock
mock_tool = Mock()
mock_tool.get_discovery_manifest.return_value = ToolResult(
output=None,
success=False,
error="Discovery failed"
)
mock_tool_class.return_value = mock_tool
# Execute and verify exception
with pytest.raises(Exception, match="Discovery failed"):
get_agent_discovery_manifest()
class TestToolInitialization:
"""Test tool initialization and dependencies."""
def test_tool_initialization_default(self):
"""Test tool initialization with default parameters."""
tool = SwarmScoreTool()
assert "api.swarmsync.ai" in tool.api_base_url
assert not tool.api_base_url.endswith('/')
def test_tool_initialization_custom_url(self):
"""Test tool initialization with custom URL."""
custom_url = "https://custom.api.com/v1/swarmscore/"
tool = SwarmScoreTool(api_base_url=custom_url)
assert tool.api_base_url == "https://custom.api.com/v1/swarmscore"
@patch('praisonai_tools.tools.swarmscore_tool.requests', None)
def test_tool_initialization_missing_requests(self):
"""Test tool initialization without requests library."""
with pytest.raises(ImportError, match="requests is required"):
SwarmScoreTool()
class TestIntegrationScenarios:
"""Test realistic integration scenarios."""
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_full_workflow_success(self, mock_requests):
"""Test complete workflow: load -> verify -> discover."""
tool = SwarmScoreTool()
# Setup mock responses
load_data = {
"score": 90,
"tier": "platinum",
"verify_payload": {"signature": "test123"}
}
verify_data = {"verified": True}
manifest_data = {"agents": ["agent1"]}
mock_responses = [
Mock(json=lambda: load_data, raise_for_status=lambda: None),
Mock(json=lambda: verify_data, raise_for_status=lambda: None),
Mock(json=lambda: manifest_data, raise_for_status=lambda: None)
]
# Configure mock to return different responses for different calls
mock_requests.exceptions = real_requests.exceptions
mock_requests.get.side_effect = [mock_responses[0], mock_responses[2]]
mock_requests.post.return_value = mock_responses[1]
# Execute workflow
load_result = tool.load_swarmscore("test-agent")
assert load_result.success is True
assert load_result.output["score"] == 90
verify_result = tool.verify_swarmscore(load_data["verify_payload"])
assert verify_result.success is True
assert verify_result.output["verified"] is True
manifest_result = tool.get_discovery_manifest()
assert manifest_result.success is True
assert "agent1" in manifest_result.output["agents"]
@patch('praisonai_tools.tools.swarmscore_tool.requests')
def test_error_recovery(self, mock_requests):
"""Test graceful error handling and recovery."""
tool = SwarmScoreTool()
# First call fails, second succeeds
mock_requests.exceptions = real_requests.exceptions
mock_requests.get.side_effect = [
real_requests.exceptions.RequestException("Network timeout"),
Mock(
json=lambda: {"score": 80},
raise_for_status=lambda: None
)
]
# First attempt fails
result1 = tool.load_swarmscore("test-agent")
assert result1.success is False
assert "Network timeout" in result1.error
# Second attempt succeeds
result2 = tool.load_swarmscore("test-agent")
assert result2.success is True
assert result2.output["score"] == 80