Snowflake + Pydantic AI: governed agents on your data Snowflake and Pydantic AI have launched a native integration, adding SnowflakeModel and SnowflakeProvider to Pydantic AI, enabling governed AI agents to run on Snowflake data via the Snowflake Cortex Inference API. The integration routes all model families through Cortex's OpenAI-compatible endpoint, requiring only two environment variables, and supports structured output, tool calling, and extended thinking. This is a guest post written by Priya Joseph , Sr. Data Cloud Architect at Snowflake. Co-authored by Douwe Maan , lead developer of Pydantic AI. Pydantic AI now has a native Snowflake provider, with the addition of SnowflakeModel and SnowflakeProvider . Users choose Snowflake for secure, governed, trusted enterprise experience. This integration brings Pydantic AI https://pydantic.dev/docs/ai/overview/ natively into Snowflake's secure perimeter. Snowflake users can now get Pydantic's built-in data validation and type safety combined with Snowflake's enterprise governance. Running a governed AI agent against your Snowflake data is simple: python import logfire from pydantic ai import Agent logfire.configure logfire.instrument pydantic ai agent = Agent 'snowflake:claude-sonnet-5' result = agent.run sync 'Summarize Q2 churn trends' The two logfire lines are optional, and worth it. With Pydantic AI instrumented https://pydantic.dev/docs/logfire/integrations/llms/pydanticai/ , every run in the rest of this post lands on one trace: the model call, the validated output, and any tool calls in between. The examples below assume they're in place. Two environment variables SNOWFLAKE ACCOUNT and SNOWFLAKE TOKEN are all the configuration needed. Everything else, including auth, routing, and governance, is handled inside the secure Snowflake perimeter. What is Snowflake Cortex Inference? Cortex Inference is a fully managed REST API that serves Claude, GPT, Llama, Mistral, DeepSeek, Grok xAI , and Snowflake's own models, all from inside your Snowflake account. Data never leaves the Snowflake security perimeter. That matters if you're in a regulated space like finance or healthcare. The interesting design choice: rather than building a separate adapter per model family, everything routes through Cortex's OpenAI-compatible Chat Completions endpoint /api/v2/cortex/v1/chat/completions . That single API surface covers tool calling, structured output json schema , image input, prompt caching, and reasoning, so one integration covers the full feature surface. See the Cortex Inference documentation https://docs.snowflake.com/en/user-guide/snowflake-cortex/cortex-rest-api for model availability. An extended example Here’s an extended example from the biology domain that showcases the power of Pydantic AI with Snowflake Cortex: Structured output structured-output-deseq2-results for DESeq2 gene expression, with validated Ensembl IDs and significance testing Tool calling tool-calling-with-validated-parameters with BLAST search parameter validation Nested models nested-models-variant-annotation for variant annotations Extended thinking extended-thinking-claude for protein analysis questions Model portability model-portability across frontier and OSS model providers, showing identical Pydantic schemas work with all models A long-running PubMed research task a-long-running-pubmed-research-task-with-polling with polling Authentication that travels The same file should run in external Python, Notebooks, Sprocs, and SPCS. python import os from pydantic ai.providers.snowflake import SnowflakeProvider Try to detect if we're running inside Snowflake Notebook, Sproc, SiS try: from snowflake.snowpark.context import get active session session = get active session SNOWFLAKE ACCOUNT = session.get current account SNOWFLAKE TOKEN = session.connection.rest. token print " Detected Snowflake environment - using session token" except ImportError: Running externally laptop, CI/CD - use environment variables SNOWFLAKE ACCOUNT = os.environ.get 'SNOWFLAKE ACCOUNT' SNOWFLAKE TOKEN = os.environ.get 'SNOWFLAKE TOKEN' if not SNOWFLAKE ACCOUNT or not SNOWFLAKE TOKEN: raise ValueError "Missing required environment variables:\n" "SNOWFLAKE ACCOUNT: your Snowflake account identifier\n" "SNOWFLAKE TOKEN: your Personal Access Token PAT \n" "Set them with: export SNOWFLAKE ACCOUNT='...' SNOWFLAKE TOKEN='...'" print " Using environment variables for authentication" Initialize provider with explicit credentials Works in: External Python, Notebooks, Streamlit-in-Snowflake, SPCS provider = SnowflakeProvider account=SNOWFLAKE ACCOUNT, token=SNOWFLAKE TOKEN, For private connectivity PrivateLink , add custom base url,needs token as well base url='https://myorg-myaccount.privatelink.snowflakecomputing.com' Structured output DESeq2 Results The Gene model validates the shape of the answer, not just its text. An Ensembl ID that doesn't match the pattern, or a fold change outside the plausible range, fails before it reaches your code. python from typing import List, Literal import logfire from pydantic import BaseModel, Field from pydantic ai import Agent logfire.configure logfire.instrument pydantic ai class Gene BaseModel : """Type-safe gene expression result.""" id: str = Field pattern=r'^ENSG\d{11}$' validates Ensembl ID format symbol: str log2fc: float = Field ge=-10, le=10 Must be between -10 and 10 padj: float = Field gt=0, le=1 P-value 0-1 @property def is significant self - bool: return self.padj < 0.05 and abs self.log2fc 1 Simple 2-line setup agent = Agent 'snowflake:claude-sonnet-5', output type=List Gene result = agent.run sync 'DUSP1 ENSG00000120129 log2FC=2.9 padj=1.2e-10' print f'Gene: {result.data 0 .symbol}, Significant: {result.data 0 .is significant}' Tool calling with validated parameters Tool inputs are validated before the function runs, so a malformed evalue or an unknown database never reaches BLAST. class BlastParams BaseModel : """Pydantic validates tool parameters automatically.""" sequence: str = Field min length=20 database: Literal 'nr', 'nt', 'refseq protein' evalue: float = Field default=0.001, gt=0, le=1 def blast search params: BlastParams - dict: """Tool input is validated before execution.""" return {'hits': 15, 'top': f'Match in {params.database}'} agent tools = Agent 'snowflake:claude-opus-4-8', tools= blast search , system prompt='You can run BLAST searches.', Agent automatically validates and calls tool result = agent tools.run sync 'BLAST sequence ATCGATCGATCGATCGATCG against RefSeq proteins' print f'Tool result: {result.data}' Nested models Variant Annotation Output types nest, so a variant annotation comes back as a typed object graph rather than a dictionary you have to pick apart. class Variant BaseModel : rsid: str = Field pattern=r'^rs\d+$' chromosome: str position: int = Field gt=0 class Annotation BaseModel : """Nested Pydantic model.""" variant: Variant Nested gene: str consequence: Literal 'missense', 'nonsense', 'synonymous' pathogenic: bool agent nested = Agent 'snowflake:claude-sonnet-5', result type=Annotation result = agent nested.run sync 'rs429358 chr19:45411941 APOE missense pathogenic' print f'Variant: {result.data.variant.rsid} in {result.data.gene}' Extended thinking Claude Claude's extended thinking is a model setting, and the validated output type still applies. class Analysis BaseModel : finding: str confidence: float = Field ge=0, le=1 agent thinking = Agent 'snowflake:claude-opus-4-8', result type=Analysis, model settings={'thinking': {'type': 'enabled', 'budget tokens': 5000}}, result = agent thinking.run sync 'Why is BRCA2 important in DNA repair?' print f'Analysis: {result.data.finding :50 }... confidence: {result.data.confidence} ' Model portability The same schema works across models. Switching between Claude, GPT, and Llama is a change of one string. php def test model model name: str - str: """The same Pydantic schema works across ALL models.""" agent = Agent model name, result type=Gene result = agent.run sync 'FKBP5 ENSG00000096433 log2FC=3.8 padj=3.4e-15' return result.data.symbol Switch models by changing ONE string for model in 'snowflake:claude-sonnet-5', 'snowflake:gpt-5.4', 'snowflake:llama3.3-70b' : gene = test model model print f'{model}: {gene}' A long-running PubMed research task with polling Real work is rarely one call. This example polls PubMed's E-utilities in batches, then hands the abstracts to Cortex for synthesis. The Field description=... strings are load-bearing: they become part of the JSON schema the model is asked to fill in. The PubMed fetching is ordinary aiohttp and not the interesting part — the complete runnable version is in this gist https://gist.github.com/laisbsc/e4c4134d3448f4508bf654657686d4be . What matters here is the schema and the agent call: python import asyncio class ResearchSummary BaseModel : """The schema the model is asked to fill in.""" key findings: List str = Field description='3-5 major findings from the literature' research gaps: List str = Field description='Identified gaps or controversies' clinical relevance: str = Field description='Clinical and translational implications' recommended reading: List str = Field description='Top 3 PMID references' async def research topic: str, model: str = 'snowflake:claude-opus-4-8' - ResearchSummary: articles = await fetch pubmed articles topic plain aiohttp; see the gist literature = '\n\n'.join f' PMID {a.pmid} {a.title}\n{a.abstract}' for a in articles :5 agent = Agent model, output type=ResearchSummary, system prompt= 'You are a biomedical research analyst. ' 'Focus on clinical relevance and research gaps.' , result = await agent.run f'Topic: {topic}\n\nRecent literature:\n{literature}' return result.output summary = asyncio.run research 'CRISPR gene editing cancer therapy' for pmid in summary.recommended reading: guaranteed List str print f'https://pubmed.ncbi.nlm.nih.gov/{pmid}/' Add logfire.instrument aiohttp client next to the earlier instrument pydantic ai call and the PubMed fetches show up on the same trace as the model call, so a slow run tells you which half was slow. result.output is a validated ResearchSummary , so summary.recommended reading is a List str . No isinstance checks, no defensive .get calls, and a clear ValidationError if the model returns something else. The implementation Two classes do the work: SnowflakeProvider handles auth and routing, SnowflakeModel handles the Cortex-specific quirks. SnowflakeProvider, auth and routing python import os from pydantic ai.providers.snowflake import SnowflakeProvider SNOWFLAKE ACCOUNT = os.environ.get 'SNOWFLAKE ACCOUNT' SNOWFLAKE TOKEN = os.environ.get 'SNOWFLAKE TOKEN' provider = SnowflakeProvider account=SNOWFLAKE ACCOUNT, token=SNOWFLAKE TOKEN, PAT, OAuth token, or key-pair JWT For private connectivity PrivateLink : base url='https://myorg-myaccount.privatelink.snowflakecomputing.com', Auth uses a plain Authorization: Bearer