Agent Security 2026: Tool Access Control — Least Privilege for AI Agents

An agent with unrestricted tool access is a loaded weapon.

Last post covered prompt injection — how attackers make agents do things they shouldn’t. This post covers the other half: when the agent acts in good faith but has too much power. A bug in tool routing, an overly permissive MCP server, or a missing parameter validation can cause more damage than any injection attack.

In production, tool access control follows the same principle that’s guided software security for 50 years: least privilege. Give the agent exactly the tools it needs, nothing more. Scope each tool to the minimum set of actions. Validate every call. Rotate credentials aggressively.

The Problem: Agents Are Too Powerful#

1
# Bad: One MCP server with everything
2
agent = Agent(
3
    mcp_servers=[{
4
        "name": "super-mcp",
5
        "command": "node",
6
        "args": ["./servers/all-in-one.js"],
7
        # Exposes 30 tools: read/write DB, manage users, deploy code,
8
        # send emails, modify billing, access admin panels...
9
    }]
10
)

The agent has access to 30 tools. It uses 5 of them regularly. The other 25 are attack surface — any one of them can be invoked by an injection attack or a routing error.

The Threat Model#

1
Agent gets compromised (injection)
2
    │
3
    ▼
4
Agent calls dangerous tool
5
    │
6
    ▼
7
Tool executes with the MCP server's credentials
8
    │
9
    ▼
10
Blast radius = whatever the MCP server's credentials allow

The blast radius depends entirely on how you scope your MCP servers and their credentials.

Principle 1: One MCP Server, One Responsibility#

1
# Good: Split MCP servers by responsibility
2
agent = Agent(
3
    mcp_servers=[
4
        {
5
            "name": "read-only-db",
6
            "command": "node",
7
            "args": ["./servers/db-reader.js"],
8
            "env": {
9
                "DATABASE_URL": os.getenv("READ_ONLY_DB_URL"),
10
                # Read-only credentials. No write access.
11
            }
12
        },
13
        {
14
            "name": "github-pr",
15
            "command": "node",
16
            "args": ["./servers/github-limited.js"],
17
            "env": {
18
                "GITHUB_TOKEN": os.getenv("GITHUB_PR_TOKEN"),
19
                # Token scoped to: pull requests only
20
                # Not the full GitHub token
21
            }
22
        },
23
        {
24
            "name": "search-docs",
25
            "command": "npx",
26
            "args": ["-y", "@internal/docs-mcp"],
27
            # No credentials needed. Public docs only.
28
        }
29
    ]
30
)

Each MCP server runs with its own credentials, scoped to its specific responsibility. If the agent is compromised, the attacker only gets access to read-only DB and PR operations — not deployment, not billing, not admin.

MCP Server Credential Scoping Table#

MCP Server	Minimum Credential	What You’d Normally Use	Risk Difference
Database	Read-only user, scoped to specific tables	Full admin user	Critical vs read-only
GitHub API	Fine-grained PAT: `contents:read`, `pulls:read`	Classic PAT with full `repo` scope	Full repo vs read-only
File system	Specific directory only	Home directory	Full home vs sandboxed
Email	Send-only API key with rate limit	Admin SMTP credentials	Full inbox vs send-only
Cloud provider	Single-service, read-only role	Admin role with full access	Everything vs one service
Slack	Channel-scoped bot token	Full workspace token	One channel vs everything

Principle 2: Scope Tools Within MCP Servers#

Even within a single MCP server, not every tool should be available to every agent.

Tool-Level Scoping#

1
# MCP server with scoped tool registration
2
class DatabaseMCPServer:
3
    def __init__(self, access_level: str):
4
        self.access_level = access_level  # "readonly" | "write" | "admin"
5

6
    def register_tools(self, mcp):
7
        # All agents can read
8
        @mcp.tool()
9
        async def query_database(sql: str) -> list[dict]:
10
            """Execute SELECT queries only."""
11
            if not sql.strip().upper().startswith("SELECT"):
12
                raise PermissionError("Only SELECT queries allowed")
13
            return await self.execute(sql)
14

15
        @mcp.tool()
16
        async def get_table_schema(table: str) -> dict:
17
            """Get table structure."""
18
            return await self.get_schema(table)
19

20
        # Only agents with write access can modify
21
        if self.access_level in ("write", "admin"):
22
            @mcp.tool()
23
            async def insert_record(table: str, data: dict) -> int:
24
                """Insert a new record. Limited to non-sensitive tables."""
25
                if table in ("users", "payments", "audit_log"):
26
                    raise PermissionError(f"Cannot insert into {table}")
27
                return await self.insert(table, data)
28

29
        # Only admin agents can delete
30
        if self.access_level == "admin":
31
            @mcp.tool()
32
            async def delete_records(table: str, where: dict) -> int:
33
                """Delete records. Requires admin access."""
34
                return await self.delete(table, where)

Tool Routing Based on Agent Role#

1
# Agent routing to different MCP instances
2
class MCPRouter:
3
    """Route agents to the right MCP server instance based on their role."""
4

5
    def __init__(self):
6
        self.readonly_db = DatabaseMCPServer("readonly")
7
        self.write_db = DatabaseMCPServer("write")
8
        self.admin_db = DatabaseMCPServer("admin")
9

10
    def get_server_for_agent(self, agent_role: str):
11
        if agent_role == "support":
12
            return self.readonly_db  # Read-only for support agents
13
        elif agent_role == "engineer":
14
            return self.write_db     # Can write to non-sensitive tables
15
        elif agent_role == "admin":
16
            return self.admin_db     # Full access (used rarely)
17
        else:
18
            return self.readonly_db  # Default: read-only

Principle 3: Validate Every Tool Call at Runtime#

Tool definitions are static. Tool calls are dynamic. Validate at call time.

Parameter-Level Validation#

1
class ToolValidator:
2
    """Validate tool parameters before execution."""
3

4
    TOOL_RULES = {
5
        "query_database": {
6
            "allowed_schemas": ["public"],
7
            "forbidden_tables": ["secrets", "credentials", "audit_log"],
8
            "max_results": 1000,
9
            "allow_write": False,
10
        },
11
        "send_email": {
12
            "allowed_domains": ["company.com", "client.org"],
13
            "max_recipients": 10,
14
            "require_approval_threshold": 100,  # Approve if > 100 recipients
15
            "forbidden_patterns": [r'@competitor\.com', r'@.*\.ru$'],
16
        },
17
        "deploy_service": {
18
            "allowed_services": ["staging-api", "staging-worker"],
19
            "forbidden_services": ["production-api", "production-worker"],
20
            "require_human_approval": True,
21
        },
22
        "file_write": {
23
            "allowed_directories": ["/app/uploads", "/app/tmp"],
24
            "forbidden_extensions": [".sh", ".exe", ".bat", ".py"],
25
            "max_file_size_mb": 10,
26
        },
27
    }
28

29
    def validate(self, tool_name: str, params: dict) -> ValidationResult:
30
        if tool_name not in self.TOOL_RULES:
31
            return ValidationResult(False, f"Unknown tool: {tool_name}")
32

33
        rules = self.TOOL_RULES[tool_name]
34

35
        if "allowed_schemas" in rules:
36
            schema = params.get("schema", "public")
37
            if schema not in rules["allowed_schemas"]:
38
                return ValidationResult(False, f"Schema '{schema}' not allowed")
39

40
        if "forbidden_tables" in rules:
41
            table = params.get("table", "")
42
            if any(t in table for t in rules["forbidden_tables"]):
43
                return ValidationResult(False, f"Table '{table}' is restricted")
44

45
        if "max_results" in rules:
46
            limit = params.get("limit", rules["max_results"])
47
            if limit > rules["max_results"]:
48
                return ValidationResult(False, f"Max results: {rules['max_results']}")
49

50
        if "require_human_approval" in rules and rules["require_human_approval"]:
51
            return ValidationResult(True, requires_approval=True)
52

53
        return ValidationResult(True)

Context-Aware Validation#

1
class ContextAwareValidator:
2
    """Validate tool calls based on conversation context."""
3

4
    def validate(self, tool_name: str, params: dict, context: ToolContext) -> ValidationResult:
5
        # Check if this tool call is expected based on conversation history
6
        if not self.is_expected_tool(tool_name, context):
7
            return ValidationResult(False, "Unexpected tool call in this context")
8

9
        # Check if the agent has already called a dangerous tool recently
10
        if self.is_suspicious_sequence(tool_name, context):
11
            return ValidationResult(False, "Suspicious tool sequence detected")
12

13
        # Check rate: too many tool calls in short time?
14
        if self.exceeds_rate_limit(context.agent_id, tool_name):
15
            return ValidationResult(False, "Rate limit exceeded for this tool")
16

17
        return ValidationResult(True)
18

19
    def is_expected_tool(self, tool_name: str, context: ToolContext) -> bool:
20
        """A tool call should make sense given the conversation."""
21
        last_tool = context.last_tool_call
22
        if tool_name == "delete_user" and last_tool != "verify_identity":
23
            return False  # Must verify identity before deletion
24
        if tool_name == "deploy_production" and last_tool != "run_tests":
25
            return False  # Must run tests before deployment
26
        return True
27

28
    def is_suspicious_sequence(self, tool_name: str, context: ToolContext) -> bool:
29
        """Detect rapid escalation of tool permissions."""
30
        recent_tools = context.recent_tool_calls(5)
31
        dangerous_tools = {"delete_*", "drop_*", "deploy_*", "admin_*"}
32
        dangerous_count = sum(1 for t in recent_tools if any(
33
            fnmatch.fnmatch(t, pattern) for pattern in dangerous_tools
34
        ))
35
        return dangerous_count >= 3  # 3+ dangerous calls in last 5 = suspicious

Principle 4: Credential Management — Never Hardcode, Always Rotate#

1
import os
2
from datetime import datetime, timedelta
3
from cryptography.fernet import Fernet
4

5
class CredentialManager:
6
    """Manage credentials for MCP servers with auto-rotation."""
7

8
    def __init__(self):
9
        self.credentials = self.load_from_vault()
10
        self.rotation_schedule = {
11
            "database": timedelta(hours=1),       # 1 hour for DB
12
            "github": timedelta(hours=2),          # 2 hours for GitHub
13
            "cloud_api": timedelta(minutes=30),    # 30 min for cloud
14
            "internal_api": timedelta(days=1),     # 1 day for internal
15
        }
16

17
    def get_credential(self, service: str, agent_id: str, session_id: str) -> str:
18
        """Get a time-limited credential for this specific session."""
19
        credential = self.credentials[service]
20

21
        # Generate scoped, short-lived token
22
        if service == "database":
23
            return self.generate_db_token(
24
                username=f"agent_{agent_id[:8]}",
25
                database="readonly",
26
                ttl=self.rotation_schedule["database"],
27
            )
28
        elif service == "github":
29
            return self.generate_github_token(
30
                permissions=["contents:read", "pulls:read"],
31
                ttl=self.rotation_schedule["github"],
32
                session_id=session_id,  # Trackable to session
33
            )
34
        # ...
35

36
    def revoke_session_credentials(self, session_id: str):
37
        """Revoke all credentials issued for a session."""
38
        for token in self.issued_tokens[session_id]:
39
            self.revoke_token(token)

Principle 5: Human-in-the-Loop for Dangerous Actions#

Some tools should never be called without human approval.

1
class HumanApprovalGate:
2
    """Require human approval for high-risk tool calls."""
3

4
    APPROVAL_REQUIRED_TOOLS = [
5
        "deploy_production",
6
        "delete_user_account",
7
        "modify_billing",
8
        "grant_admin_access",
9
        "bulk_email",
10
        "database_migration",
11
    ]
12

13
    async def request_approval(
14
        self,
15
        tool_name: str,
16
        params: dict,
17
        agent_reasoning: str,
18
    ) -> bool:
19
        """Request human approval for a tool call."""
20

21
        approval_request = {
22
            "tool": tool_name,
23
            "params": params,
24
            "reasoning": agent_reasoning,
25
            "requested_at": datetime.utcnow().isoformat(),
26
            "expires_at": (datetime.utcnow() + timedelta(minutes=5)).isoformat(),
27
            "approval_id": uuid4().hex,
28
        }
29

30
        # Send to approval queue
31
        await self.approval_queue.send(approval_request)
32

33
        # Wait for response (with timeout)
34
        try:
35
            response = await self.approval_queue.wait_for_response(
36
                approval_request["approval_id"],
37
                timeout=300,  # 5 minutes
38
            )
39
            return response["approved"]
40
        except TimeoutError:
41
            await self.alert_team(f"Approval request expired: {tool_name}")
42
            return False
43

44
    async def log_auto_blocked(self, tool_name: str, params: dict, reason: str):
45
        """Log when a tool call is blocked without human review."""
46
        await self.audit_log.append({
47
            "event": "tool_blocked",
48
            "tool": tool_name,
49
            "params": params,
50
            "reason": reason,
51
            "timestamp": datetime.utcnow().isoformat(),
52
        })

Production Implementation: Putting It All Together#

1
class SecureToolOrchestrator:
2
    """Complete tool access control system."""
3

4
    def __init__(self):
5
        self.validator = ToolValidator()
6
        self.context_validator = ContextAwareValidator()
7
        self.credential_manager = CredentialManager()
8
        self.approval_gate = HumanApprovalGate()
9
        self.audit_log = AuditLogger()
10

11
    async def execute_tool(
12
        self,
13
        tool_name: str,
14
        params: dict,
15
        agent_id: str,
16
        session_id: str,
17
        context: ToolContext,
18
    ) -> ToolResult:
19

20
        # 1. Static validation (tool rules)
21
        validation = self.validator.validate(tool_name, params)
22
        if not validation.passed:
23
            await self.audit_log.blocked_call(
24
                agent_id, tool_name, params, f"Static validation: {validation.reason}"
25
            )
26
            return ToolResult(error=validation.reason)
27

28
        # 2. Context-aware validation
29
        context_check = self.context_validator.validate(tool_name, params, context)
30
        if not context_check.passed:
31
            await self.audit_log.blocked_call(
32
                agent_id, tool_name, params, f"Context validation: {context_check.reason}"
33
            )
34
            return ToolResult(error=context_check.reason)
35

36
        # 3. Human approval for dangerous tools
37
        if tool_name in self.approval_gate.APPROVAL_REQUIRED_TOOLS:
38
            approved = await self.approval_gate.request_approval(
39
                tool_name, params, context.agent_reasoning
40
            )
41
            if not approved:
42
                return ToolResult(error="Requires human approval")
43

44
        # 4. Get scoped credentials for this session
45
        credentials = self.credential_manager.get_credential(
46
            tool_name.split("_")[0], agent_id, session_id
47
        )
48

49
        # 5. Execute with monitoring
50
        start_time = time.time()
51
        try:
52
            result = await self.execute_with_credentials(
53
                tool_name, params, credentials
54
            )
55

56
            # 6. Validate output
57
            output_validation = self.validate_output(tool_name, result)
58
            if not output_validation.passed:
59
                await self.audit_log.blocked_output(
60
                    agent_id, tool_name, output_validation.reason
61
                )
62
                return ToolResult(error="Output validation failed")
63

64
            # 7. Log successful call
65
            await self.audit_log.tool_call(
66
                agent_id=agent_id,
67
                tool=tool_name,
68
                params=params,
69
                duration_ms=(time.time() - start_time) * 1000,
70
                success=True,
71
            )
72

73
            return ToolResult(data=result)
74

75
        except Exception as e:
76
            await self.audit_log.tool_call(
77
                agent_id=agent_id,
78
                tool=tool_name,
79
                params=params,
80
                duration_ms=(time.time() - start_time) * 1000,
81
                success=False,
82
                error=str(e),
83
            )
84
            return ToolResult(error=str(e))

Testing Tool Access Control#

1
TOOL_ACCESS_TEST_CASES = [
2
    # Agent role, tool call, expected result
3
    ("support", "query_database", True),       # Support can read
4
    ("support", "insert_record", False),        # Support can't write
5
    ("support", "delete_records", False),       # Support can't delete
6
    ("engineer", "query_database", True),       # Engineer can read
7
    ("engineer", "insert_record", True),        # Engineer can write (limited)
8
    ("engineer", "delete_records", False),      # Engineer can't delete
9
    ("admin", "query_database", True),          # Admin can read
10
    ("admin", "delete_records", True),          # Admin can delete
11
    # Suspicious sequences
12
    ("engineer", ["query_users", "delete_users"], False),  # Must verify before delete
13
    # Parameter validation
14
    ("support", "query_database", {"table": "secrets"}, False),  # Can't query secrets
15
]
16

17
def test_tool_access_control(orchestrator):
18
    results = []
19
    for role, tool, expected in TOOL_ACCESS_TEST_CASES:
20
        context = ToolContext(agent_role=role)
21
        if isinstance(tool, list):
22
            # Sequential call test
23
            for t in tool[:1]:
24
                orchestrator.execute_tool(t, {}, role, "test", context)
25
            result = orchestrator.execute_tool(tool[-1], {}, role, "test", context)
26
        elif isinstance(tool, str):
27
            result = orchestrator.execute_tool(tool, {}, role, "test", context)
28

29
        passed = result.success == expected
30
        results.append({"role": role, "tool": tool, "expected": expected, "passed": passed})
31

32
    return results

Production Checklist#

Next in the Series#

Post	Topic
1	Prompt Injection & Defense
2	Tool Access Control (this)
3	MCP Server Security
4	Agent Auditing & Compliance
5	Production Security Patterns

Series: Agent Security 2026 — Production Patterns. Post 2: Tool Access Control — least privilege for AI agents.