Bastio
Agent Security

Agent Security SDK (Beta)

TypeScript and Python SDKs for easy AI agent security integration.

Agent Security SDK (Beta)

The Agent Security SDK is currently in beta. We're accepting early adopters for testing and feedback.

The Bastio Agent Security SDK provides a simple, type-safe way to integrate AI agent security into your applications. Available for TypeScript and Python.

Join the Beta

We're looking for developers building AI agents to help us refine our SDKs.

Benefits of joining the beta:

  • Early access to SDK releases
  • Direct support from our engineering team
  • Influence on SDK design and features
  • Free Pro tier during beta period

Available SDKs

TypeScript SDK

npm install @bastio/agent-security  # Coming soon

Features:

  • Full TypeScript type definitions
  • Async/await support
  • Automatic request signing
  • Built-in retry logic
  • OpenAI and Anthropic format support

Python SDK

pip install bastio-agent-security  # Coming soon

Features:

  • Full type hints (Python 3.9+)
  • Async/await with asyncio
  • Sync API for simple use cases
  • Automatic request signing
  • Pydantic models

Preview: TypeScript SDK

import { BastioAgentSecurity } from '@bastio/agent-security';

// Initialize
const bastio = new BastioAgentSecurity({
  apiKey: process.env.BASTIO_API_KEY,
  proxyId: 'your-proxy-id',
  agentId: process.env.BASTIO_AGENT_ID,  // Optional
  agentPrivateKey: process.env.BASTIO_AGENT_KEY,  // Optional
});

// Validate a tool call
const result = await bastio.validateToolCall({
  sessionId: 'session_123',
  toolCall: {
    id: 'call_abc',
    type: 'function',
    function: {
      name: 'execute_shell',
      arguments: JSON.stringify({ command: 'ls -la' }),
    },
  },
});

// Handle the result
if (result.action === 'allow') {
  // Safe to execute
  const output = await executeShell('ls -la');

  // Scan the output before returning to agent
  const scanResult = await bastio.scanContent({
    content: output,
    contentType: 'tool_output',
    toolName: 'execute_shell',
  });

  return scanResult.safeContent;

} else if (result.action === 'require_approval') {
  // Wait for human approval
  const approved = await bastio.waitForApproval(result.approvalId, {
    timeoutMs: 300000,  // 5 minutes
    pollIntervalMs: 5000,
  });

  if (approved) {
    return await executeShell('ls -la');
  } else {
    return 'Operation was not approved.';
  }

} else {
  // Blocked
  return `Operation blocked: ${result.message}`;
}

Preview: Python SDK

from bastio import AgentSecurity, ToolCall

# Initialize
bastio = AgentSecurity(
    api_key=os.environ["BASTIO_API_KEY"],
    proxy_id="your-proxy-id",
    agent_id=os.environ.get("BASTIO_AGENT_ID"),
    agent_private_key=os.environ.get("BASTIO_AGENT_KEY"),
)

# Validate a tool call
result = await bastio.validate_tool_call(
    session_id="session_123",
    tool_call=ToolCall(
        id="call_abc",
        type="function",
        name="execute_shell",
        arguments={"command": "ls -la"}
    )
)

# Handle the result
match result.action:
    case "allow":
        output = await execute_shell("ls -la")

        # Scan output
        scan_result = await bastio.scan_content(
            content=output,
            content_type="tool_output",
            tool_name="execute_shell"
        )
        return scan_result.safe_content

    case "require_approval":
        approved = await bastio.wait_for_approval(
            approval_id=result.approval_id,
            timeout_seconds=300
        )
        if approved:
            return await execute_shell("ls -la")
        return "Operation not approved"

    case "block":
        return f"Blocked: {result.message}"

SDK Features

Automatic Tool Call Validation

The SDK can automatically intercept and validate tool calls:

// Wrap your tool executor
const secureExecutor = bastio.wrapToolExecutor(async (toolCall) => {
  // Your actual tool execution logic
  return await executeTool(toolCall);
});

// Now all tool calls are automatically validated
const result = await secureExecutor({
  name: 'read_file',
  arguments: { path: '/etc/passwd' }
});
// If blocked, throws BastioSecurityError
// If requires approval, waits for approval
// If allowed, executes and returns result
# Decorator for automatic validation
@bastio.validate_tool
async def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()

# Now all calls are validated
try:
    content = await read_file("/etc/passwd")
except BastioSecurityError as e:
    print(f"Blocked: {e.message}")

Session Management

Track tool calls across a conversation:

// Create a session
const session = bastio.createSession({
  userId: 'user_123',
  metadata: { conversation_id: 'conv_456' }
});

// All tool calls in this session are tracked together
await session.validateToolCall(toolCall1);
await session.validateToolCall(toolCall2);

// Chain analysis works across the session

Approval Handling

Simple approval flow management:

# Async approval waiting
result = await bastio.validate_tool_call(session_id, tool_call)

if result.action == "require_approval":
    # Wait with timeout and callback
    approved = await bastio.wait_for_approval(
        result.approval_id,
        timeout_seconds=300,
        on_pending=lambda: notify_user("Waiting for approval...")
    )

Content Scanning

Scan retrieved content for threats:

// Scan web search results
const searchResults = await webSearch(query);
const safeResults = await bastio.scanContent({
  content: searchResults,
  contentType: 'tool_output',
  toolName: 'web_search'
});

// Scan RAG documents
const documents = await vectorSearch(embedding);
const safeDocuments = await bastio.scanMultiple(
  documents.map(doc => ({
    content: doc.text,
    contentType: 'rag_document',
    metadata: { source: doc.source }
  }))
);

Error Handling

Typed errors for easy handling:

import { BastioError, SecurityBlockedError, ApprovalTimeoutError } from '@bastio/agent-security';

try {
  await bastio.validateToolCall(toolCall);
} catch (error) {
  if (error instanceof SecurityBlockedError) {
    console.log('Threats:', error.threats);
    console.log('Risk score:', error.riskScore);
  } else if (error instanceof ApprovalTimeoutError) {
    console.log('Approval timed out after', error.timeoutMs, 'ms');
  }
}

Integration Examples

OpenAI Function Calling

import OpenAI from 'openai';
import { BastioAgentSecurity } from '@bastio/agent-security';

const openai = new OpenAI();
const bastio = new BastioAgentSecurity({ ... });

async function processWithSecurity(messages: Message[]) {
  const response = await openai.chat.completions.create({
    model: 'gpt-4',
    messages,
    tools: myTools,
  });

  const toolCalls = response.choices[0].message.tool_calls;

  for (const toolCall of toolCalls) {
    // Validate with Bastio
    const validation = await bastio.validateToolCall({
      sessionId: conversationId,
      toolCall: {
        id: toolCall.id,
        type: 'function',
        function: toolCall.function,
      },
    });

    if (validation.action === 'allow') {
      // Execute tool
      const output = await executeFunction(toolCall.function);

      // Scan output
      const scanResult = await bastio.scanContent({
        content: JSON.stringify(output),
        toolName: toolCall.function.name,
      });

      // Continue conversation with safe output
    }
  }
}

LangChain Integration

from langchain.agents import AgentExecutor
from bastio import AgentSecurity

bastio = AgentSecurity(...)

class SecureTool(BaseTool):
    """Wrapper that adds Bastio validation to any tool."""

    def __init__(self, tool: BaseTool, bastio: AgentSecurity):
        self.tool = tool
        self.bastio = bastio

    async def _arun(self, **kwargs) -> str:
        # Validate
        result = await self.bastio.validate_tool_call(
            session_id=get_current_session(),
            tool_call=ToolCall(
                name=self.tool.name,
                arguments=kwargs
            )
        )

        if result.action != "allow":
            raise ToolSecurityError(result.message)

        # Execute
        output = await self.tool._arun(**kwargs)

        # Scan output
        scan = await self.bastio.scan_content(output)
        return scan.safe_content

# Wrap all tools
secure_tools = [SecureTool(t, bastio) for t in original_tools]
agent = AgentExecutor(tools=secure_tools, ...)

Current API Alternative

While the SDK is in beta, you can use our REST API directly. See:

Roadmap

Q1 2025:

  • TypeScript SDK beta release
  • Python SDK beta release
  • Basic tool validation and content scanning

Q2 2025:

  • Framework integrations (LangChain, LlamaIndex)
  • Streaming support
  • Batch operations

Q3 2025:

  • General availability
  • Additional language SDKs (Go, Java)
  • Advanced features (custom patterns, webhooks)

Stay Updated

Join the beta waitlist to receive updates on SDK development:

Next Steps