mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-09-30 11:51:46 +02:00
Merge pull request #303 from Carlos-Projects/feat/agno-mcp-integration
feat: add MCP+Agno integration docs and report chart tests
This commit is contained in:
2 files changed
+233
No files matched your search
@@ -0,0 +1,156 @@
|
|||||||
|
# MCP + Agno Integration
|
||||||
|
|
||||||
|
This guide shows how to use Agentic Security's MCP server with [Agno](https://docs.agno.com/tools/mcp) agents.
|
||||||
|
|
||||||
|
## Setup
|
||||||
|
|
||||||
|
Install Agentic Security with optional Agno support:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install agno
|
||||||
|
```
|
||||||
|
|
||||||
|
## Starting the MCP Server
|
||||||
|
|
||||||
|
Start the Agentic Security MCP server:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python -m agentic_security.mcp.main
|
||||||
|
```
|
||||||
|
|
||||||
|
For production, use the stdio transport (default with FastMCP):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python agentic_security/mcp/main.py
|
||||||
|
```
|
||||||
|
|
||||||
|
## Examples
|
||||||
|
|
||||||
|
### Basic Verification with Agno
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from agno.agent import Agent
|
||||||
|
from agno.tools.mcp import MCPTools
|
||||||
|
|
||||||
|
from agentic_security.mcp.main import mcp
|
||||||
|
|
||||||
|
|
||||||
|
async def verify_llm_spec():
|
||||||
|
# Connect to Agentic Security's MCP server via stdio
|
||||||
|
mcp_tools = MCPTools(
|
||||||
|
command="python",
|
||||||
|
args=["agentic_security/mcp/main.py"],
|
||||||
|
)
|
||||||
|
await mcp_tools.connect()
|
||||||
|
|
||||||
|
try:
|
||||||
|
agent = Agent(
|
||||||
|
tools=[mcp_tools],
|
||||||
|
instructions=[
|
||||||
|
"You are a security testing assistant.",
|
||||||
|
"Use verify_llm to test LLM specifications for vulnerabilities.",
|
||||||
|
"Present results clearly with risk levels.",
|
||||||
|
],
|
||||||
|
markdown=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
await agent.aprint_response(
|
||||||
|
"Verify this LLM spec: openai/gpt-4",
|
||||||
|
stream=True,
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
await mcp_tools.close()
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(verify_llm_spec())
|
||||||
|
```
|
||||||
|
|
||||||
|
### Running a Security Scan
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from agno.agent import Agent
|
||||||
|
from agno.tools.mcp import MCPTools
|
||||||
|
|
||||||
|
|
||||||
|
async def run_security_scan():
|
||||||
|
mcp_tools = MCPTools(
|
||||||
|
command="python",
|
||||||
|
args=["agentic_security/mcp/main.py"],
|
||||||
|
)
|
||||||
|
await mcp_tools.connect()
|
||||||
|
|
||||||
|
try:
|
||||||
|
agent = Agent(
|
||||||
|
tools=[mcp_tools],
|
||||||
|
instructions=[
|
||||||
|
"You are an LLM security scanning assistant.",
|
||||||
|
"Use start_scan to initiate security scans on LLM endpoints.",
|
||||||
|
"Use get_data_config to check available scan configurations.",
|
||||||
|
"Report findings with severity levels.",
|
||||||
|
],
|
||||||
|
markdown=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
await agent.aprint_response(
|
||||||
|
"Run a security scan on openai/gpt-4 with max budget 100",
|
||||||
|
stream=True,
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
await mcp_tools.close()
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(run_security_scan())
|
||||||
|
```
|
||||||
|
|
||||||
|
### Streamable HTTP Transport
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from agno.agent import Agent
|
||||||
|
from agno.tools.mcp import MCPTools
|
||||||
|
|
||||||
|
|
||||||
|
async def run_http_transport():
|
||||||
|
mcp_tools = MCPTools(
|
||||||
|
transport="streamable-http",
|
||||||
|
url="http://0.0.0.0:8718/mcp",
|
||||||
|
)
|
||||||
|
await mcp_tools.connect()
|
||||||
|
|
||||||
|
try:
|
||||||
|
agent = Agent(
|
||||||
|
tools=[mcp_tools],
|
||||||
|
markdown=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
await agent.aprint_response(
|
||||||
|
"List available security scan templates",
|
||||||
|
stream=True,
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
await mcp_tools.close()
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(run_http_transport())
|
||||||
|
```
|
||||||
|
|
||||||
|
## Available Tools
|
||||||
|
|
||||||
|
| Tool | Description |
|
||||||
|
|---|---|
|
||||||
|
| `verify_llm` | Verify an LLM model specification |
|
||||||
|
| `start_scan` | Start an LLM security scan |
|
||||||
|
| `stop_scan` | Stop an ongoing scan |
|
||||||
|
| `get_data_config` | Retrieve data configuration |
|
||||||
|
| `get_spec_templates` | Retrieve LLM specification templates |
|
||||||
|
|
||||||
|
## Notes
|
||||||
|
|
||||||
|
- The stdio transport is recommended for local development
|
||||||
|
- For production deployments, use the streamable-http transport
|
||||||
|
- Always call `mcp_tools.close()` to clean up connections
|
||||||
@@ -0,0 +1,77 @@
|
|||||||
|
import io
|
||||||
|
|
||||||
|
import matplotlib
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
matplotlib.use("Agg")
|
||||||
|
|
||||||
|
from agentic_security.report_chart import (
|
||||||
|
_generate_identifiers,
|
||||||
|
generate_identifiers,
|
||||||
|
plot_security_report,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestGenerateIdentifiers:
|
||||||
|
def test_single_row(self):
|
||||||
|
data = type("DF", (), {"__len__": lambda s: 1})()
|
||||||
|
result = _generate_identifiers(data)
|
||||||
|
assert result == ["A1"]
|
||||||
|
|
||||||
|
def test_multiple_rows(self):
|
||||||
|
data = type("DF", (), {"__len__": lambda s: 5})()
|
||||||
|
result = _generate_identifiers(data)
|
||||||
|
assert result == ["A1", "A2", "A3", "A4", "A5"]
|
||||||
|
|
||||||
|
def test_alphabet_wraparound(self):
|
||||||
|
data = type("DF", (), {"__len__": lambda s: 27})()
|
||||||
|
result = _generate_identifiers(data)
|
||||||
|
assert result[0] == "A1"
|
||||||
|
assert result[25] == "A26"
|
||||||
|
assert result[26] == "B1"
|
||||||
|
|
||||||
|
def test_empty_dataframe(self):
|
||||||
|
data = type("DF", (), {"__len__": lambda s: 0})()
|
||||||
|
result = _generate_identifiers(data)
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
def test_public_generate_identifiers(self):
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
df = pd.DataFrame({"a": [1, 2, 3]})
|
||||||
|
result = generate_identifiers(df)
|
||||||
|
assert result == ["A1", "A2", "A3"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestPlotSecurityReport:
|
||||||
|
def test_returns_bytesio(self):
|
||||||
|
table_data = [
|
||||||
|
{"module": "test", "failureRate": 10.0, "tokens": 100},
|
||||||
|
]
|
||||||
|
result = plot_security_report(table_data)
|
||||||
|
assert isinstance(result, io.BytesIO)
|
||||||
|
|
||||||
|
def test_multiple_modules(self):
|
||||||
|
table_data = [
|
||||||
|
{"module": "mod_a", "failureRate": 5.0, "tokens": 50},
|
||||||
|
{"module": "mod_b", "failureRate": 15.0, "tokens": 200},
|
||||||
|
{"module": "mod_c", "failureRate": 25.0, "tokens": 500},
|
||||||
|
]
|
||||||
|
result = plot_security_report(table_data)
|
||||||
|
assert result.getvalue() != b""
|
||||||
|
|
||||||
|
def test_handles_empty_data(self):
|
||||||
|
result = plot_security_report([])
|
||||||
|
assert isinstance(result, io.BytesIO)
|
||||||
|
|
||||||
|
def test_handles_missing_keys(self):
|
||||||
|
table_data = [{"module": "test"}]
|
||||||
|
result = plot_security_report(table_data)
|
||||||
|
assert isinstance(result, io.BytesIO)
|
||||||
|
|
||||||
|
def test_handles_none_values(self):
|
||||||
|
table_data = [
|
||||||
|
{"module": "test", "failureRate": None, "tokens": None},
|
||||||
|
]
|
||||||
|
result = plot_security_report(table_data)
|
||||||
|
assert isinstance(result, io.BytesIO)
|
||||||
Reference in new issue
Block a user