Technical Design Document (TDD): Gemini URL Context Integration

Document ID: TDD-2026-0204-004
Date: February 4, 2026
Version: 1.0
Status: Draft

---

1. Implementation Architecture

1.1 Technology Stack

Layer	Technology	Rationale
API Client	Raw HTTP (httpx/aiohttp)	REST API over SDK per presenter recommendation
Caching	FoundationDB	Existing Coditect state backbone
Audit Storage	FoundationDB (immutable keyspace)	Compliance requirement; 7-year retention
Agent Framework	Coditect Multi-Agent Orchestrator	Existing orchestration infrastructure
Model Routing	Coditect Model Router	Cost optimization via Haiku/Sonnet/Opus equivalent
Monitoring	OpenTelemetry → Coditect Observability	Unified platform metrics

1.2 Module Structure

coditect/
├── web_intelligence/
│   ├── __init__.py
│   ├── adapter.py           # URLContextAdapter - Gemini API wrapper
│   ├── cache.py             # ContentCache - FoundationDB caching
│   ├── audit.py             # AuditTrail - compliance audit records
│   ├── models.py            # Data models (requests, responses, metadata)
│   ├── router.py            # ModelRouter - intelligent model selection
│   ├── sanitizer.py         # ContentSanitizer - input/output sanitization
│   ├── validator.py         # URLValidator - allowlisting, format checks
│   ├── circuit_breaker.py   # CircuitBreaker - failure protection
│   └── tools/
│       ├── __init__.py
│       ├── url_context.py   # Coditect tool interface for URL Context
│       ├── web_research.py  # Combined Search + URL Context tool
│       └── compliance_research.py  # Compliance-specific research tool
├── agents/
│   ├── researcher.py        # Enhanced with web intelligence tools
│   └── compliance.py        # Enhanced with compliance research tools
└── tests/
    ├── test_adapter.py
    ├── test_cache.py
    ├── test_audit.py
    ├── test_integration.py
    └── test_compliance.py

---

2. Core Implementation

2.1 URL Context Adapter (REST API)

import httpx
import hashlib
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
import asyncio

@dataclass
class GeminiURLContextAdapter:
    """Direct REST API adapter for Gemini URL Context.
    
    Uses REST API over SDK to:
    - Reduce dependency complexity
    - Avoid SDK version churn
    - Maintain direct control over request/response handling
    """
    
    api_key: str
    base_url: str = "https://generativelanguage.googleapis.com/v1beta"
    default_model: str = "gemini-2.5-flash"
    timeout: float = 30.0
    max_retries: int = 3
    
    _client: httpx.AsyncClient = field(init=False, repr=False)
    
    def __post_init__(self):
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(self.timeout),
            headers={"Content-Type": "application/json"}
        )
    
    async def fetch_and_analyze(
        self,
        urls: List[str],
        prompt: str,
        model: Optional[str] = None,
        include_search: bool = False,
    ) -> Dict[str, Any]:
        """Fetch URL content and analyze with Gemini."""
        
        model = model or self.default_model
        endpoint = f"{self.base_url}/models/{model}:generateContent"
        
        # Build tool configuration
        tools = [{"url_context": {}}]
        if include_search:
            tools.append({"google_search": {}})
        
        # Build URL references into prompt
        url_refs = "\n".join(f"- {url}" for url in urls)
        full_prompt = f"{prompt}\n\nSource URLs:\n{url_refs}"
        
        payload = {
            "contents": [{"parts": [{"text": full_prompt}]}],
            "tools": tools
        }
        
        # Execute with retry
        for attempt in range(self.max_retries):
            try:
                response = await self._client.post(
                    endpoint,
                    json=payload,
                    params={"key": self.api_key}
                )
                
                if response.status_code == 429:
                    wait = 2 ** attempt
                    await asyncio.sleep(wait)
                    continue
                
                response.raise_for_status()
                return self._parse_response(response.json())
                
            except httpx.TimeoutException:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)
        
        raise RuntimeError(f"Failed after {self.max_retries} retries")
    
    def _parse_response(self, data: Dict) -> Dict[str, Any]:
        """Extract content and metadata from Gemini response."""
        candidate = data["candidates"][0]
        
        # Extract text content
        text_parts = [
            part["text"] for part in candidate["content"]["parts"]
            if "text" in part
        ]
        
        # Extract URL metadata
        url_metadata = []
        if "urlContextMetadata" in candidate:
            for meta in candidate["urlContextMetadata"].get("urlMetadata", []):
                url_metadata.append({
                    "url": meta.get("retrievedUrl", ""),
                    "status": meta.get("urlRetrievalStatus", "UNKNOWN")
                })
        
        # Extract token usage
        usage = data.get("usageMetadata", {})
        
        return {
            "content": "\n".join(text_parts),
            "url_metadata": url_metadata,
            "tokens": {
                "input": usage.get("promptTokenCount", 0),
                "output": usage.get("candidatesTokenCount", 0),
                "total": usage.get("totalTokenCount", 0)
            },
            "timestamp": datetime.utcnow().isoformat()
        }

2.2 Model Routing for URL Context Tasks

class URLContextModelRouter:
    """Route URL Context requests to optimal Gemini model."""
    
    ROUTING_RULES = {
        # Task type → (model, rationale)
        "research": ("gemini-2.5-flash", "Cost-efficient for general research"),
        "compliance": ("gemini-2.5-pro", "Higher accuracy for regulatory docs"),
        "pdf_analysis": ("gemini-2.5-pro", "Better visual document understanding"),
        "api_docs": ("gemini-2.5-flash", "Structured content; Flash sufficient"),
        "news": ("gemini-2.5-flash", "Simple extraction; cost priority"),
        "code_review": ("gemini-2.5-flash", "Code is structured; Flash handles well"),
        "architecture": ("gemini-2.5-pro", "Complex reasoning required"),
    }
    
    def select_model(
        self,
        task_type: str,
        regulatory: bool = False,
        content_types: List[str] = None
    ) -> str:
        if regulatory:
            return "gemini-2.5-pro"
        
        if content_types and "pdf" in content_types:
            return "gemini-2.5-pro"
        
        model, _ = self.ROUTING_RULES.get(
            task_type, 
            ("gemini-2.5-flash", "default")
        )
        return model

2.3 FoundationDB Cache Implementation

import fdb
import json
import time
from enum import Enum

fdb.api_version(710)

class ContentCategory(Enum):
    REGULATORY = "regulatory"
    DOCUMENTATION = "documentation"
    NEWS = "news"
    STATIC = "static"
    VOLATILE = "volatile"

class URLContentCache:
    """FoundationDB-backed cache for URL Context responses."""
    
    TTL_SECONDS = {
        ContentCategory.REGULATORY: 86400,     # 24 hours
        ContentCategory.DOCUMENTATION: 21600,   # 6 hours
        ContentCategory.NEWS: 3600,             # 1 hour
        ContentCategory.STATIC: 604800,         # 7 days
        ContentCategory.VOLATILE: 1800,         # 30 minutes
    }
    
    def __init__(self, db: fdb.Database):
        self.db = db
        self.cache_space = fdb.directory.create_or_open(
            db, ('coditect', 'web_intelligence', 'cache')
        )
    
    @fdb.transactional
    def get(self, tr, url_hash: str) -> Optional[Dict]:
        """Get cached content if not expired."""
        key = self.cache_space.pack((url_hash,))
        value = tr[key]
        
        if value is None:
            return None
        
        entry = json.loads(value)
        if time.time() > entry.get("expires_at", 0):
            del tr[key]  # Clean up expired entry
            return None
        
        return entry
    
    @fdb.transactional
    def put(self, tr, url_hash: str, content: Dict, category: ContentCategory):
        """Store content with category-based TTL."""
        ttl = self.TTL_SECONDS[category]
        entry = {
            **content,
            "cached_at": time.time(),
            "expires_at": time.time() + ttl,
            "category": category.value
        }
        key = self.cache_space.pack((url_hash,))
        tr[key] = json.dumps(entry).encode()
    
    @staticmethod
    def hash_url(url: str) -> str:
        return hashlib.sha256(url.encode()).hexdigest()[:32]

2.4 Compliance Audit Trail

import uuid
from dataclasses import dataclass, asdict

@dataclass(frozen=True)
class ComplianceAuditRecord:
    """Immutable audit record for regulatory compliance.
    
    Satisfies:
    - FDA 21 CFR Part 11.10(e): Audit trail
    - HIPAA §164.312(b): Audit controls
    - SOC2 CC8.1: Change management
    """
    
    record_id: str
    timestamp: str
    agent_id: str
    agent_role: str
    task_description: str
    urls_requested: List[str]
    urls_retrieved: List[str]
    retrieval_statuses: Dict[str, str]
    model_used: str
    tokens_input: int
    tokens_output: int
    prompt_hash: str
    response_hash: str
    cache_hit: bool
    regulatory_context: str
    content_hashes: Dict[str, str]  # URL → SHA-256 of content
    
    @classmethod
    def create(
        cls,
        agent_id: str,
        agent_role: str,
        task_description: str,
        request: Dict,
        response: Dict,
        regulatory_context: str = ""
    ) -> "ComplianceAuditRecord":
        return cls(
            record_id=str(uuid.uuid4()),
            timestamp=datetime.utcnow().isoformat() + "Z",
            agent_id=agent_id,
            agent_role=agent_role,
            task_description=task_description,
            urls_requested=request.get("urls", []),
            urls_retrieved=[m["url"] for m in response.get("url_metadata", [])],
            retrieval_statuses={
                m["url"]: m["status"] 
                for m in response.get("url_metadata", [])
            },
            model_used=request.get("model", "unknown"),
            tokens_input=response.get("tokens", {}).get("input", 0),
            tokens_output=response.get("tokens", {}).get("output", 0),
            prompt_hash=hashlib.sha256(
                request.get("prompt", "").encode()
            ).hexdigest(),
            response_hash=hashlib.sha256(
                response.get("content", "").encode()
            ).hexdigest(),
            cache_hit=response.get("cached", False),
            regulatory_context=regulatory_context,
            content_hashes={}
        )

class AuditTrailStore:
    """FoundationDB-backed immutable audit trail."""
    
    def __init__(self, db: fdb.Database):
        self.db = db
        self.audit_space = fdb.directory.create_or_open(
            db, ('coditect', 'web_intelligence', 'audit')
        )
    
    @fdb.transactional
    def record(self, tr, audit: ComplianceAuditRecord):
        """Write immutable audit record."""
        key = self.audit_space.pack((
            audit.timestamp,
            audit.record_id
        ))
        tr[key] = json.dumps(asdict(audit)).encode()
    
    @fdb.transactional
    def query_by_timerange(
        self, tr, start: str, end: str
    ) -> List[ComplianceAuditRecord]:
        """Query audit records within time range."""
        start_key = self.audit_space.pack((start,))
        end_key = self.audit_space.pack((end,))
        
        records = []
        for kv in tr.get_range(start_key, end_key):
            data = json.loads(kv.value)
            records.append(ComplianceAuditRecord(**data))
        return records

---

3. Coditect Tool Interface

3.1 URL Context Tool for Agent Framework

from coditect.tools import BaseTool, ToolResult

class URLContextTool(BaseTool):
    """Coditect tool interface for Gemini URL Context.
    
    Enables agents to fetch and analyze web content autonomously
    with built-in caching, audit trails, and compliance support.
    """
    
    name = "url_context"
    description = """Fetch and analyze content from web URLs.
    Supports HTML, PDF, images, JSON, XML, CSV.
    Use for: documentation research, compliance checks, API analysis.
    Limits: 20 URLs/request, 34MB/URL, public URLs only."""
    
    def __init__(
        self,
        adapter: GeminiURLContextAdapter,
        cache: URLContentCache,
        audit: AuditTrailStore,
        model_router: URLContextModelRouter
    ):
        self.adapter = adapter
        self.cache = cache
        self.audit = audit
        self.router = model_router
    
    async def execute(
        self,
        urls: List[str],
        analysis_prompt: str,
        task_type: str = "research",
        compliance_mode: bool = False,
        regulatory_context: str = "",
        force_refresh: bool = False
    ) -> ToolResult:
        # Validate URLs
        validated = [u for u in urls if await self._validate_url(u, compliance_mode)]
        if not validated:
            return ToolResult(
                success=False,
                error="No valid URLs provided"
            )
        
        # Check cache (unless force refresh)
        if not force_refresh:
            cached = await self._check_cache(validated, analysis_prompt)
            if cached:
                return ToolResult(success=True, data=cached)
        
        # Route to optimal model
        model = self.router.select_model(
            task_type=task_type,
            regulatory=compliance_mode
        )
        
        # Execute API call
        response = await self.adapter.fetch_and_analyze(
            urls=validated,
            prompt=analysis_prompt,
            model=model,
            include_search=(task_type == "research")
        )
        
        # Verify retrieval success
        failed_urls = [
            m["url"] for m in response["url_metadata"]
            if m["status"] != "URL_RETRIEVAL_STATUS_SUCCESS"
        ]
        if failed_urls:
            response["warnings"] = f"Failed to retrieve: {failed_urls}"
        
        # Cache response
        await self._update_cache(validated, analysis_prompt, response, task_type)
        
        # Record audit trail
        if compliance_mode:
            audit_record = ComplianceAuditRecord.create(
                agent_id=self.current_agent_id,
                agent_role=self.current_agent_role,
                task_description=analysis_prompt,
                request={"urls": validated, "model": model, "prompt": analysis_prompt},
                response=response,
                regulatory_context=regulatory_context
            )
            self.audit.record(self.db, audit_record)
        
        return ToolResult(success=True, data=response)

3.2 Compliance Research Tool

class ComplianceResearchTool(BaseTool):
    """Specialized tool for regulatory document research.
    
    Combines URL Context with domain-specific validation
    for FDA, HIPAA, and SOC2 compliance workflows.
    """
    
    name = "compliance_research"
    description = """Research regulatory requirements from authoritative sources.
    Automatically enforces URL allowlisting, audit trails, and content verification.
    Supported frameworks: FDA 21 CFR Part 11, HIPAA, SOC2."""
    
    AUTHORITATIVE_SOURCES = {
        "fda_21cfr11": [
            "https://www.fda.gov/regulatory-information/search-fda-guidance-documents",
            "https://www.ecfr.gov/current/title-21/chapter-I/subchapter-A/part-11",
        ],
        "hipaa": [
            "https://www.hhs.gov/hipaa/index.html",
            "https://www.hhs.gov/hipaa/for-professionals/security/index.html",
        ],
        "soc2": [
            "https://www.aicpa-cima.com/topic/audit-assurance/audit-and-assurance-greater-than-soc-2",
        ],
    }
    
    async def research_requirement(
        self,
        framework: str,
        requirement_query: str,
        additional_urls: List[str] = None
    ) -> ToolResult:
        """Research specific compliance requirement with full audit trail."""
        
        urls = self.AUTHORITATIVE_SOURCES.get(framework, [])
        if additional_urls:
            urls.extend(additional_urls)
        
        return await self.url_context_tool.execute(
            urls=urls,
            analysis_prompt=f"""
            Framework: {framework}
            Query: {requirement_query}
            
            Extract specific requirements, control objectives, and implementation
            guidance. Cite the exact section/paragraph for each requirement.
            Flag any areas of ambiguity requiring human review.
            """,
            task_type="compliance",
            compliance_mode=True,
            regulatory_context=framework
        )

---

4. Performance Specifications

Metric	Target	Measurement
Cache hit ratio	> 50% (steady state)	FoundationDB metrics
API response latency (P50)	< 5s	End-to-end timer
API response latency (P99)	< 15s	End-to-end timer
URL retrieval success rate	> 95%	Metadata parsing
Audit record write latency	< 10ms	FoundationDB timer
Token cost per research task	< $0.05 (Flash) / < $0.20 (Pro)	Token counter

---

5. Migration Path

Phase 1: Adapter + Cache (Week 1-2)

• Implement GeminiURLContextAdapter
• Implement FoundationDB ContentCache
• Unit tests and integration tests

Phase 2: Agent Integration (Week 3-4)

• Build URLContextTool and ComplianceResearchTool
• Integrate into Researcher and Compliance agents
• Integration testing with real URLs

Phase 3: Audit Trail + Compliance (Week 5-6)

• Implement AuditTrailStore
• Compliance testing with regulatory documents
• Human checkpoint integration

Phase 4: Production Hardening (Week 7-8)

• Circuit breaker implementation
• Monitoring and alerting setup
• Performance benchmarking
• Documentation and training

---

Document maintained by Coditect Engineering
Review cycle: Bi-weekly during implementation