# A practical guide to building secure, governed AI agents that can access your data without compromising security or compliance.

---

Most AI agents need access to structured data—your CRMs, product databases, data warehouses. The naive approach is to give agents direct database credentials and let them query whatever they need. This creates several critical problems:

## Security Risks:
- Agents can query any table, any column, any row
- No control over what sensitive data gets exposed
- Risk of accidental data leaks or malicious queries
- Difficult to implement row-level security

## Compliance Challenges:
- No audit trail of what data was accessed
- Can't demonstrate data governance to auditors
- Hard to meet regulatory requirements (GDPR, HIPAA, SOC 2)
- No way to prove agents only see approved data

## Operational Issues:
- Can't update data schemas without breaking agents
- Difficult to optimize queries (agents write ad-hoc SQL)
- No way to join data across multiple sources
- Performance issues from inefficient queries

## The Solution: A Layered Architecture

Instead of direct access, we need a layered architecture where agents interact with data through controlled, governed interfaces. This gives you security, compliance, and operational control while still enabling powerful AI capabilities.

---

## The 5-Layer Architecture

### Layer 1: Data Sources

Your raw data repositories—completely off-limits to AI agents.

**What Goes Here:**
- Production databases (PostgreSQL, MySQL, MongoDB)
- Data warehouses (Snowflake, BigQuery, Redshift)
- SaaS platforms (Salesforce, HubSpot, Stripe)
- APIs and data lakes

**Key Principle:** Agents never get credentials to these sources. They can't query them directly, ever.

**Implementation:**
- Store connection credentials securely (encrypted, in secrets management)
- Use read-only credentials where possible
- Implement network-level isolation if needed
- Monitor all connections for anomalies

---

### Layer 2: Data Governance & Security (The Critical Boundary)

This is where you define exactly what data agents can access. Materialized SQL views act as controlled windows into your data.

**What Goes Here:**
- Materialized views that join data across sources
- Security filters (row-level and column-level)
- Data transformations and aggregations
- Purpose-built views for specific agent use cases

**Key Principle:** This is the **only** access point for agents. No exceptions.

#### Creating Agent Views

Agent views are different from traditional database views. They're purpose-built for AI agent consumption:

**1. Cross-Source Joins**

```sql
CREATE MATERIALIZED VIEW customer_health_comprehensive AS
SELECT
    -- From Salesforce
    s.account_id,
    s.account_name,
    s.account_owner,
    s.contract_value,
    s.renewal_date,

-- From Product Database
    p.active_users,
    p.feature_adoption_score,
    p.days_since_last_login,
    p.usage_trend,

-- From Snowflake Analytics
    sf.total_support_tickets,
    sf.avg_ticket_resolution_days,
    sf.revenue_last_quarter,
    sf.churn_risk_score,
    sf.churn_risk_indicators,

-- Calculated fields
    CASE
        WHEN sf.churn_risk_score > 0.7 THEN 'High Risk'
        WHEN sf.churn_risk_score > 0.4 THEN 'Medium Risk'
        ELSE 'Low Risk'
    END as risk_category,

-- Timestamp for freshness tracking
    CURRENT_TIMESTAMP as view_refreshed_at

FROM salesforce_accounts s
LEFT JOIN product_usage_metrics p
    ON s.account_id = p.account_id
LEFT JOIN snowflake_customer_analytics sf
    ON s.account_id = sf.account_id
WHERE s.account_status = 'Active'
  AND s.account_type IN ('Enterprise', 'Mid-Market');  -- Row-level filtering
```

**2. Column-Level Security**

Exclude sensitive columns from agent access:

```sql
CREATE MATERIALIZED VIEW customer_safe_view AS
SELECT
    customer_id,
    customer_name,
    email,
    account_status,
    -- Explicitly exclude: ssn, credit_card, internal_notes
    subscription_tier,
    signup_date
FROM customers
WHERE account_status = 'active';
```

**3. Row-Level Security**

Filter rows based on security policies:

```sql
CREATE MATERIALIZED VIEW my_territory_customers AS
SELECT
    account_id,
    account_name,
    revenue,
    health_score
FROM all_customers
WHERE account_owner = CURRENT_USER()  -- Only see own accounts
  AND region IN (
      SELECT region
      FROM user_territories
      WHERE user_id = CURRENT_USER()
  );
```

**4. Data Masking**

Mask sensitive data while keeping it useful:

```sql
CREATE MATERIALIZED VIEW customer_anonymized AS
SELECT
    customer_id,
    -- Mask email: john.doe@company.com -> j***@company.com
    REGEXP_REPLACE(email, '^([^@]{1,3}).*@', '\1***@') as email_masked,
    -- Hash PII for analytics
    MD5(phone_number) as phone_hash,
    account_status,
    subscription_tier
FROM customers;
```

**5. Aggregations and Pre-computation**

Pre-compute expensive aggregations:

```sql
CREATE MATERIALIZED VIEW customer_metrics_daily AS
SELECT
    customer_id,
    DATE_TRUNC('day', event_timestamp) as date,
    COUNT(*) as event_count,
    COUNT(DISTINCT user_id) as active_users,
    AVG(session_duration) as avg_session_duration,
    SUM(revenue) as daily_revenue
FROM events
GROUP BY customer_id, DATE_TRUNC('day', event_timestamp);
```

#### Materialization Strategy

**Why Materialize Views:**
- Performance: Pre-computed results are fast
- Isolation: Agents query materialized data, not live sources
- Stability: Schema changes in sources don't break agents
- Cost: Reduces load on production databases

**Refresh Strategy:**

```sql
-- Daily refresh for most views
REFRESH MATERIALIZED VIEW customer_health_comprehensive;

-- Or incremental refresh for large datasets
REFRESH MATERIALIZED VIEW CONCURRENTLY customer_metrics_daily;
```

**Best Practices:**
- Refresh on a schedule (daily, hourly) based on data freshness needs
- Use incremental refreshes for large datasets
- Monitor refresh times and optimize slow queries
- Version your views (use schema migrations)

---

### Layer 3: MCP Tool Interface

Model Context Protocol (MCP) tools expose your views as callable functions that agents can use. Each tool is a self-documenting function with parameters, validation, and policy checks.

**What Goes Here:**
- Function definitions with names and descriptions
- Parameter schemas and validation
- SQL queries that reference agent views
- Policy checks (authentication, permissions, row-level security)

**Key Principle:** Tools are the agent's API. They should be discoverable, well-documented, and secure.

#### Building MCP Tools

**1. Basic Tool Structure**

```json
{
  "name": "get_customer_health",
  "description": "Retrieves comprehensive health data for a specific customer account including usage metrics, support tickets, revenue, and churn risk indicators. Use this when users ask about customer status, health scores, or account details.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "account_name": {
        "type": "string",
        "description": "The name of the customer account (e.g., 'Acme Corp')"
      }
    },
    "required": ["account_name"]
  },
  "query": "SELECT * FROM customer_health_comprehensive WHERE account_name = $1",
  "policies": [
    "authenticated",
    "role:customer_success",
    "row_level_security:territory_match"
  ]
}
```

**2. Tool with Multiple Parameters**

```json
{
  "name": "identify_at_risk_customers",
  "description": "Identifies customer accounts at high or medium risk of churning, ordered by risk score. Returns account details, risk indicators, and key metrics.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "risk_level": {
        "type": "string",
        "enum": ["High", "Medium", "All"],
        "description": "Filter by risk level. Defaults to 'All' if not specified.",
        "default": "All"
      },
      "limit": {
        "type": "integer",
        "description": "Maximum number of results to return",
        "minimum": 1,
        "maximum": 100,
        "default": 20
      },
      "min_revenue": {
        "type": "number",
        "description": "Minimum contract value to include",
        "minimum": 0
      }
    }
  },
  "query": "SELECT * FROM customer_health_comprehensive WHERE risk_category = CASE WHEN $1 = 'All' THEN risk_category ELSE $1 END AND contract_value >= COALESCE($3, 0) ORDER BY churn_risk_score DESC LIMIT $2",
  "policies": [
    "authenticated",
    "role:customer_success OR role:manager"
  ]
}
```

**3. Tool with Date Range Filtering**

```json
{
  "name": "analyze_customer_trends",
  "description": "Analyzes usage trends and support patterns for accounts with declining engagement over a specified time period.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "start_date": {
        "type": "string",
        "format": "date",
        "description": "Start date for analysis (YYYY-MM-DD)"
      },
      "end_date": {
        "type": "string",
        "format": "date",
        "description": "End date for analysis (YYYY-MM-DD)"
      }
    },
    "required": ["start_date", "end_date"]
  },
  "query": "SELECT * FROM customer_health_comprehensive WHERE usage_trend = 'declining' AND view_refreshed_at BETWEEN $1::timestamp AND $2::timestamp",
  "policies": [
    "authenticated",
    "role:manager"
  ]
}
```

#### Policy Checks

Policies enforce security at the tool level:

**Authentication Policy:**

```python
def check_authentication(request):
    token = request.headers.get('X-Pylar-API-Key')
    if not token or not validate_token(token):
        raise UnauthorizedError("Invalid or missing authentication token")
    return get_user_from_token(token)
```

**Role-Based Access Control:**

```python
def check_role(user, required_roles):
    user_roles = get_user_roles(user.id)
    if not any(role in user_roles for role in required_roles):
        raise ForbiddenError(f"User must have one of: {required_roles}")
```

**Row-Level Security:**

```python
def apply_row_level_security(user, query):
    # Modify query to filter by user's territory
    territory_filter = f"account_owner = '{user.id}' OR region IN ({get_user_regions(user.id)})"
    return f"{query} AND {territory_filter}"
```

#### Tool Descriptions Matter

The description field is critical—it's what the LLM uses to decide when to call your tool. Be specific:

**Bad:**

```json
"description": "Gets customer data"
```

**Good:**

```json
"description": "Retrieves comprehensive health data for a specific customer account including usage metrics, support tickets, revenue, and churn risk indicators. Use this when users ask about customer status, health scores, account details, or want to understand why a customer might be at risk."
```

**Best Practices:**
- Describe what the tool does and when to use it
- Include example use cases in the description
- Be specific about what data is returned
- Mention any important limitations or filters

---

### Layer 4: AI Agent Layer

Your LLM-powered agent (LangGraph, Cursor, n8n, etc.) that interprets user queries, selects appropriate tools, and synthesizes responses.

**What Goes Here:**
- Your agent framework (LangGraph, LangChain, etc.)
- LLM configuration (model, temperature, etc.)
- Tool discovery and selection logic
- Response synthesis and formatting

**Key Principle:** Agents are stateless consumers of MCP tools. They don't know about your data sources—only the tools available to them.

#### Agent Implementation Example

**LangGraph Agent:**

```python
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import requests

class PylarAgent:
    def __init__(self, pylar_api_key, pylar_endpoint):
        self.api_key = pylar_api_key
        self.endpoint = pylar_endpoint
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        self.tools = self._discover_tools()

def _discover_tools(self):
        response = requests.get(
            f"{self.endpoint}/tools",
            headers={"X-Pylar-API-Key": self.api_key}
        )
        return response.json()["tools"]

def _select_tool(self, user_query, tools):
        tool_descriptions = "\n".join([
            f"- {t['name']}: {t['description']}" for t in tools
        ])

prompt = f"""Given the user query, select the most appropriate tool.

Available tools:
{tool_descriptions}

User query: {user_query}

Return only the tool name."""

response = self.llm.invoke([HumanMessage(content=prompt)])
        selected_tool = response.content.strip()
        return next(t for t in tools if t["name"] == selected_tool)

def _call_tool(self, tool, parameters):
        response = requests.post(
            f"{self.endpoint}/tools/{tool['name']}/execute",
            headers={"X-Pylar-API-Key": self.api_key},
            json={"parameters": parameters}
        )
        return response.json()

def process_query(self, user_query, user_context=None):
        tool = self._select_tool(user_query, self.tools)
        parameters = self._extract_parameters(user_query, tool)
        tool_result = self._call_tool(tool, parameters)
        response = self._synthesize_response(user_query, tool_result)
        return response
```

**Tool Selection with Function Calling:**

Modern LLMs support function calling, which makes tool selection easier:

```python
def get_tool_schema(tool):
    return {
        "type": "function",
        "function": {
            "name": tool["name"],
            "description": tool["description"],
            "parameters": tool["inputSchema"]
        }
    }

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": user_query}],
    tools=[get_tool_schema(t) for t in available_tools],
    tool_choice="auto"
)
```

---

### Layer 5: User Interface

End users interact with your agent through chat interfaces, APIs, or applications.

**What Goes Here:**
- Chat UI (web, Slack, Teams)
- API endpoints
- Mobile apps
- Voice interfaces

**Key Principle:** Users don't need to know about the architecture. They just ask questions and get answers.

---

## Complete Flow: From Query to Response

Here's what happens when a user asks a question:

**User Query:** "Which customers are at high risk?"

1. **Agent Processing:**
   - LLM interprets the query
   - Reviews available MCP tools
   - Selects `identify_at_risk_customers` based on tool description
2. **Tool Invocation:**
   - Agent calls Pylar MCP endpoint
   - Includes authentication header
   - Passes parameters: `{risk_level: "High"}`
3. **Policy Validation:**
   - Validates authentication token
   - Checks user role
   - Validates parameter format
   - Applies row-level security if needed
4. **Query Execution:**
   - Executes SQL against agent view
   - Materialized view returns pre-computed results
5. **Response Formatting:**
   - Formats results as JSON
   - Logs query for audit trail
6. **Response Synthesis:**
   - Synthesizes natural language response
7. **User Receives Answer:**
   - "I found 12 customers at high risk. The top 3 are: Acme Corp (risk score: 0.85), TechStart (0.78), Global Solutions (0.72). Would you like detailed analysis for any of these?"

---

## Implementation Guide

### Step 1: Connect Your Data Sources

Use Pylar (or similar platform):
1. Navigate to Integrations section
2. Add data source connections:
   - Salesforce: OAuth connection
   - PostgreSQL: Connection string (use read-only user)
   - Snowflake: Credentials (encrypted storage)
3. Test connections
4. Verify read access

**Security Checklist:**
- [ ]  Use read-only credentials where possible
- [ ]  Store credentials encrypted
- [ ]  Use network isolation if needed
- [ ]  Monitor connection logs

### Step 2: Create Agent Views

In Pylar SQL IDE (or your database):
1. Write SQL query joining your data sources
2. Add security filters
3. Test the query
4. Save as materialized view
5. Set refresh schedule

**Example View Creation:**

```sql
CREATE MATERIALIZED VIEW customer_health_comprehensive AS
    -- ... (full query from above)
```

### Step 3: Build MCP Tools

In Pylar (or your MCP server):
1. Create new MCP tool
2. Define function name and description
3. Specify input schema (parameters)
4. Write SQL query referencing your view
5. Configure policy checks
6. Test the tool

**Example Tool Definition:**
```json
{
  "name": "get_customer_health",
  "description": "Retrieves comprehensive health data for a customer account.",
  "inputSchema": {
      "type": "object",
      "properties": {
          "account_name": {
              "type": "string"
          }
      },
      "required": ["account_name"]
  },
  "query": "SELECT * FROM customer_health_comprehensive WHERE account_name = $1",
  "policies": ["authenticated"]
}
```

### Step 4: Test in Agent Playground

Before publishing, test your tools:
1. Open agent playground
2. Try sample queries:
   - "What's the health status of Acme Corp?"
3. Review observability dashboard:
   - Monitor query execution times and policy check results

### Step 5: Publish to Agent Builder

**Generate credentials:**
- API key: `pylar_sk_xxxxx`
- Endpoint URL: `https://api.pylar.ai/mcp`

### Step 6: Monitor and Iterate

Set up monitoring:
- Query logs
- Performance metrics
- Error tracking
- Usage analytics

---

## Security Best Practices
### 1. Principle of Least Privilege
- Agents should only access the minimum data needed.

### 2. Row-Level Security
- Filter data based on user context.

### 3. Column-Level Security
- Exclude sensitive columns.

### 4. Data Masking
- Mask sensitive data.

### 5. Audit Everything
- Log all queries.

### 6. Regular Reviews
- Regularly audit access logs and update security policies as needed.

---

## Performance Optimization
### 1. Materialize Expensive Queries
### 2. Index Your Views
### 3. Incremental Refreshes
### 4. Query Optimization

---

## Common Patterns
### Pattern 1: Customer Health Dashboard
### Pattern 2: Sales Pipeline Analysis
### Pattern 3: Product Usage Analytics

---

## Troubleshooting
### Agent Selects Wrong Tool
### Slow Query Performance
### Policy Check Failures
### Data Freshness Issues

---

## Conclusion
This 5-layer architecture provides a secure, governed way to connect AI agents to your data stack. The key takeaways are:
- Agents never access raw databases—only through views.
- Views are materialized for performance and isolation.
- Every query is logged for compliance.

**Next Steps:**  
1. Start with one data source and one view  
2. Build a simple MCP tool  
3. Test in an agent playground  
4. Iterate based on feedback  
5. Scale to more sources and use cases

---

*This architecture pattern is implemented in Pylar, but the concepts apply to any system connecting AI agents to structured data.*