为 AgentCore 网页搜索加上域名与发布日期过滤

import boto3

gateway_client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")

# Create a Web Search target pinned to version 1.2.0 with admin-level domain filtering
target = gateway_client.create_gateway_target(
    gatewayIdentifier="your-gateway-id",
    name="web-search-filtered",
    targetConfiguration={
        "mcp": {
            "connector": {
                "source": {"connectorId": "web-search", "version": "1.2.0"},
                "configurations": [
                    {
                        "name": "WebSearch",
                        "parameterValues": {
                            "domainFilter": {
                                "include": [
                                    "approved-wire-1.com",
                                    "approved-wire-2.com",
                                    "sec.gov",
                                    "investor.gov",
                                ],
                                "exclude": ["unreliable-source.net"],
                            }
                        },
                    }
                ],
            }
        }
    },
    credentialProviderConfigurations=[
        {"credentialProviderType": "GATEWAY_IAM_ROLE"}
    ],
)

print(f"Target ID: {target['targetId']}")
print(f"Status: {target['status']}")

{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "tools/call",
    "params": {
        "name": "WebSearch",
        "arguments": {
            "query": "latest SEC enforcement actions 2026",
            "filters": {
                "domainFilter": {
                    "include": ["sec.gov"],
                    "exclude": []
                },
                "publishedDateFilter": {
                    "from": "2026-07-01T00:00:00Z",
                    "to": "2026-08-04T23:59:59Z"
                }
            }
        }
    }
}

import json

# Pharma research assistant: only approved regulatory sources
compliance_request = json.dumps({
    "jsonrpc": "2.0",
    "id": "req-001",
    "method": "tools/call",
    "params": {
        "name": "WebSearch",
        "arguments": {
            "query": "FDA drug approval process 2026 guidance",
            "maxResults": 10,
            "filters": {
                "domainFilter": {
                    "include": ["fda.gov", "nih.gov", "clinicaltrials.gov"]
                }
            }
        }
    }
})

# Send via your MCP client (SigV4-signed or OAuth-authenticated)
response = mcp_client.send(compliance_request)
results = json.loads(response)

# Every result is from fda.gov, nih.gov, or clinicaltrials.gov
for result in results["result"]["content"]:
    print(result["text"])

from datetime import datetime, timedelta, timezone

# Calculate the 7-day window
now = datetime.now(timezone.utc)
seven_days_ago = now - timedelta(days=7)

stock_update_request = json.dumps({
    "jsonrpc": "2.0",
    "id": "req-002",
    "method": "tools/call",
    "params": {
        "name": "WebSearch",
        "arguments": {
            "query": "semiconductor stocks latest developments",
            "maxResults": 15,
            "filters": {
                "publishedDateFilter": {
                    "from": seven_days_ago.strftime("%Y-%m-%dT%H:%M:%SZ"),
                    "to": now.strftime("%Y-%m-%dT%H:%M:%SZ")
                }
            }
        }
    }
})

response = mcp_client.send(stock_update_request)

# IR compliance agent: primary sources + current quarter only
financial_request = json.dumps({
    "jsonrpc": "2.0",
    "id": "req-003",
    "method": "tools/call",
    "params": {
        "name": "WebSearch",
        "arguments": {
            "query": "AMZN quarterly earnings report Q2 2026",
            "maxResults": 10,
            "filters": {
                "domainFilter": {
                    "include": ["sec.gov", "investor.gov"],
                    "exclude": ["example-crowd-commentary-1.com", "example-crowd-commentary-2.com"]
                },
                "publishedDateFilter": {
                    "from": "2026-04-01T00:00:00Z",
                    "to": "2026-08-04T23:59:59Z"
                }
            }
        }
    }
})

response = mcp_client.send(financial_request)

from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client

# Per-tenant domain policies (from your config database)
TENANT_POLICIES = {
    "tenant-healthcare": {
        "include": ["pubmed.ncbi.nlm.nih.gov", "who.int", "cdc.gov"],
        "exclude": ["example-consumer-health-site.com"],
    },
    "tenant-legal": {
        "include": ["law.cornell.edu", "supremecourt.gov", "uscourts.gov"],
        "exclude": [],
    },
    "tenant-finance": {
        "include": ["sec.gov", "federalreserve.gov", "example-approved-wire.com"],
        "exclude": ["example-social-forum-site.com"],
    },
}


def search_for_tenant(tenant_id: str, query: str, days_back: int = 30):
    """Execute a policy-compliant web search for a specific tenant."""
    from datetime import datetime, timedelta, timezone

    policy = TENANT_POLICIES[tenant_id]
    now = datetime.now(timezone.utc)
    date_from = (now - timedelta(days=days_back)).strftime("%Y-%m-%dT%H:%M:%SZ")

    def create_transport():
        return streamablehttp_client(
            gateway_url,
            headers={"Authorization": f"Bearer {get_token()}"},
        )

    mcp_client = MCPClient(create_transport)

    model = BedrockModel(
        model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
        max_tokens=2048,
    )

    with mcp_client:
        tools = mcp_client.list_tools_sync()
        agent = Agent(model=model, tools=tools)

        # The agent's system prompt instructs it to pass these filters
        # on every WebSearch invocation
        system_prompt = f"""You are a research assistant. When using WebSearch,
ALWAYS include these filters in your tool call:

- domainFilter.include: {policy['include']}
- domainFilter.exclude: {policy['exclude']}
- publishedDateFilter.from: {date_from}

Never search outside approved domains."""

        result = agent(query, system_prompt=system_prompt)
        return result.message


# Usage
answer = search_for_tenant(
    "tenant-healthcare",
    "What are the latest clinical trial results for GLP-1 drugs?"
)