CrewAI Integration
The signalstack-crewai package provides a SignalStackTool for CrewAI agents. Add verification as a crew task, and let your agents automatically check claims, businesses, documents, and media before acting on information.
Installation
Code
pip install signalstack-crewai
# With all extras
pip install "signalstack-crewai[async]"Basic usage
Code
from signalstack_crewai import SignalStackTool
from crewai import Agent, Task, Crew
# Create the verification tool
verify_tool = SignalStackTool(
api_key="ssk_live_YOUR_API_KEY",
min_trust_score=0.7,
)
# Create a fact-checker agent
fact_checker = Agent(
role="Fact Checker",
goal="Verify all claims and data before the crew makes decisions",
backstory="I specialize in using SignalStack to verify information across multiple sources",
tools=[verify_tool],
allow_delegation=False,
verbose=True,
)
# Create a researcher agent that depends on verified info
researcher = Agent(
role="Researcher",
goal="Gather and analyze verified information",
backstory="I only work with verified facts",
tools=[verify_tool],
)
# Tasks
verify_task = Task(
description=(
"Verify the following claim using SignalStack: "
"'The Great Wall of China is visible from space with the naked eye'"
),
expected_output="A trust score and verdict with supporting evidence",
agent=fact_checker,
)
research_task = Task(
description="Summarize the verified facts about the Great Wall of China",
expected_output="A bullet-point summary based on verified information",
agent=researcher,
)
# Create and run the crew
crew = Crew(
agents=[fact_checker, researcher],
tasks=[verify_task, research_task],
verbose=True,
)
result = crew.kickoff()
# The fact checker will call SignalStack to verify the claim
# Agent determines: Trust score 0.04 — Verdict: False
# Researcher then uses the verified result for their summarySequential verification pipeline
Code
from signalstack_crewai import SignalStackTool
from crewai import Agent, Task, Crew, Process
verify = SignalStackTool(api_key="ssk_live_...", min_trust_score=0.8)
screener = Agent(
role="Content Screener",
goal="Screen all content before it reaches the writing team",
tools=[verify],
)
writer = Agent(
role="Writer",
goal="Write accurate content based on verified facts",
)
screen_task = Task(
description="Verify this document: https://example.com/article.pdf",
expected_output="Authenticity analysis and trust score",
agent=screener,
)
write_task = Task(
description="Write a summary of the verified document",
expected_output="A one-paragraph summary",
agent=writer,
)
crew = Crew(
agents=[screener, writer],
tasks=[screen_task, write_task],
process=Process.sequential,
)
crew.kickoff()Configuration
| Option | Type | Default | Description |
|---|---|---|---|
api_key | str | SIGNALSTACK_API_KEY env | API key |
min_trust_score | float | 0.8 | Minimum acceptable trust score |
on_threshold_fail | str | "raise" | "raise" or "return_evidence" |
timeout | int | 30 | Request timeout in seconds |
max_retries | int | 3 | Max retries for transient errors |
cache_results | bool | True | Cache results in memory to avoid duplicate calls |
Tool input schema
The tool accepts a JSON string with the following fields:
| Field | Type | Required | Description |
|---|---|---|---|
action | string | Yes | Verification type: verify_claim, verify_business,verify_document, verify_media |
params | object | Yes | Endpoint-specific parameters (same as API) |
Error handling
Code
from signalstack_crewai import SignalStackTool, TrustThresholdError
tool = SignalStackTool(
api_key="ssk_live_...",
min_trust_score=0.8,
on_threshold_fail="return_evidence",
)
{`# When trust score is below threshold, the tool returns:
# {
# "error": "Trust threshold not met",
# "trust_score": 0.04,
# "threshold": 0.8,
# "evidence": [...],
# "message": "The claim is not supported by authoritative sources"
# }`}