Free lesson · GenAI Application Engineering
Build a code validator with Gemini ToolCodeExecution
You will build a CodeValidator class in processing/code_validator.py using Gemini 2.5 Flash native code execution to validate Python blocks in a sandbox. The validator wraps google.genai.Client calling client.aio.models.generate_content() with tools=[ToolCodeExecution], passing code in the prompt. From response.candidates[0].content.parts, you extract executable_code and code_execution_result, checking outcome for SUCCESS or FAILED. A ValidationResult Pydantic model captures code_block_id, executed, output, error, execution_time_ms, and outcome (Literal['success','error','timeout']). You implement validate_all(blocks: list[CodeBlock]) -> list[ValidationResult] using asyncio.gather() for concurrency. POST /api/v1/process/validate accepts CodeBlock list and returns ValidationResult list.
Course: Full-Stack GenAI Applications · Chapter 4 · Message Processing Pipeline
Free to read — no subscription required.
Introduction
When an LLM streams Python code into a chat response, you have no guarantee that the snippet actually runs — syntactically valid code still throws ImportError, references undefined names, or hangs on an infinite loop. Teams that ship unverified blocks to learners erode trust the moment a paste-and-run example breaks in front of a user. By the end of this lesson you'll be able to validate extracted Python code blocks using Gemini 2.5 Flash's native ToolCodeExecution sandbox, so every Python block your pipeline emits has been proven runnable server-side before it reaches the client.
Key Terminology
- ToolCodeExecution: The Gemini SDK tool declaration that enables server-side execution of Python code inside an isolated sandbox managed by Google, returning structured
ExecutableCodeandCodeExecutionResultparts in the model response. - CodeValidator: The stateless class in
processing/code_validator.pythat wraps agoogle.genai.Client, constructs the execution prompt, invokes Gemini with theToolCodeExecutiontool, and parses the response into aValidationResult. - ValidationResult: The Pydantic model returned by
CodeValidator.validate, carrying aValidationStatus(SUCCESS,ERROR, orTIMEOUT), captured stdout, an error message, and the exact code that was executed.
Concepts
Practical Considerations for Production Deployments
Rate limiting and batching. Gemini API calls for code execution consume the same quota as standard generation requests. If a single LLM response contains five or more Python blocks, sequential validation creates noticeable latency. Use asyncio.gather with the async variant of the client (await client.aio.models.generate_content(...)) to validate blocks concurrently. The google-genai SDK exposes client.aio for async operations, which maps directly onto the same ToolCodeExecution configuration shown above.
Sandbox limitations. Gemini's code execution sandbox supports standard library modules and a curated set of third-party packages (NumPy, pandas, matplotlib, and others in the scientific Python stack). It does not support network access, file system writes, or arbitrary pip installs. Code blocks that import unsupported modules will return OUTCOME_FAILED. Your pipeline should treat these failures as inconclusive rather than invalid — the code might be perfectly correct but simply untestable in the sandbox.
Error classification. Not all execution errors indicate bad code. Distinguish between syntax errors (which ast.parse from another goal should have already caught), runtime errors (import failures, type errors, assertion failures), and resource errors (timeout, memory exceeded). Map each category to a different UI treatment — syntax errors warrant a red indicator, runtime errors an orange warning, and resource errors a gray "could not verify" badge.
Security boundaries. Although Gemini's sandbox is isolated, the prompt you send contains user-influenced content. Sanitize the code string to remove prompt-injection attempts — for example, text that tries to instruct Gemini to ignore the execution directive and instead return fabricated success output. The temperature=0.0 setting and explicit prompt phrasing reduce but do not eliminate this risk. For high-security deployments, validate that the ExecutableCode part in the response matches the code you submitted, which the _parse_response method already supports by comparing executed against original_code.
Code Walkthrough
Architecture: Where Code Validation Fits
The following diagram shows how a Python code block flows from initial extraction through sandbox validation and back into the content assembly pipeline. Notice that the CodeValidator operates as a synchronous gate — downstream SSE emission waits for the validation verdict before including the code block in the streamed response.
Code snippet mermaid
Loading diagram...
- Line 1: Declares a Mermaid flowchart with top-down (TD) layout direction.
- Line 2: Defines the starting node where a raw LLM response is processed through Instructor extraction into a CodeBlock Pydantic model.
- Line 3: Introduces a decision diamond checking whether the extracted code block's language is Python.
- Line 4: Handles the "No" branch—if the language is not Python, the block is marked as unvalidated and passed through without execution.
- Line 5: Handles the "Yes" branch—if the language is Python, the block is sent to CodeValidator.validate for runtime verification.
- Line 6: Shows that validation is performed by calling Gemini 2.5 Flash with ToolCodeExecution, which executes the Python code in a sandboxed environment.
- Line 7: Introduces a second decision diamond that inspects the CodeExecutionResult outcome returned by Gemini.
- Line 8: On
OUTCOME_OK, the code block is marked as validated=True and the captured stdout is attached to the result. - Line 9: On
OUTCOME_ERROR, the code block is marked as validated=False and the error details are attached to the result. - Lines 10-12: All three terminal paths (validated success, validated failure, and unvalidated pass-through) converge into a single node where the block is emitted to the client as a Server-Sent Event (SSE) verified block.
This design keeps code execution entirely off your own servers. The CodeValidator class is stateless — it holds a configured google.genai.Client and a model identifier, but no execution state. Each validation call is independent, which means you can run multiple validations concurrently using asyncio.gather when a single LLM response contains several Python blocks.
Building the CodeValidator Class
The CodeValidator class lives in processing/code_validator.py and serves as the single integration point with Gemini's code execution API. It accepts a CodeBlock Pydantic model (produced by the Instructor extraction phase from another goal), constructs a prompt that instructs Gemini to execute the code, and parses the response parts into a ValidationResult model. The class uses google.genai.Client from the google-genai SDK rather than the older google-generativeai package, and it configures ToolCodeExecution as part of the tools parameter in the generate_content call. The constructor accepts an optional timeout parameter that defaults to 30 seconds — long enough for reasonable code but short enough to prevent runaway execution from blocking the pipeline.
Code snippetpython
1from google import genai 2from google.genai import types 3from pydantic import BaseModel, Field 4from enum import Enum 5 6class ValidationStatus(str, Enum): 7 SUCCESS = "success" 8 ERROR = "error" 9 TIMEOUT = "timeout" 10 11class ValidationResult(BaseModel): 12 status: ValidationStatus 13 stdout: str = Field(default="") 14 error_message: str = Field(default="") 15 executed_code: str = Field(default="") 16 17class CodeValidator: 18 def __init__(self, model: str = "gemini-2.5-flash", timeout: int = 30): 19 self.client = genai.Client() 20 self.model = model 21 self.timeout = timeout 22 self.code_execution_tool = types.Tool( 23 code_execution=types.ToolCodeExecution() 24 ) 25 26 def validate(self, code: str) -> ValidationResult: 27 fence = "```" 28 prompt = ( 29 "Execute the following Python code and report the result. " 30 "Do not modify the code. Run it exactly as provided.\n\n" 31 f"{fence}python\n{code}\n{fence}" 32 ) 33 try: 34 response = self.client.models.generate_content( 35 model=self.model, 36 contents=prompt, 37 config=types.GenerateContentConfig( 38 tools=[self.code_execution_tool], 39 temperature=0.0, 40 ), 41 ) 42 return self._parse_response(response, code) 43 except Exception as exc: 44 return ValidationResult( 45 status=ValidationStatus.TIMEOUT, 46 error_message=str(exc), 47 executed_code=code, 48 ) 49 50 def _parse_response( 51 self, response: types.GenerateContentResponse, original_code: str 52 ) -> ValidationResult: 53 stdout_parts = [] 54 error_parts = [] 55 executed = original_code 56 57 for part in response.candidates[0].content.parts: 58 if hasattr(part, "executable_code") and part.executable_code: 59 executed = part.executable_code.code or executed 60 61 if hasattr(part, "code_execution_result") and part.code_execution_result: 62 result = part.code_execution_result 63 if result.outcome == types.Outcome.OUTCOME_OK: 64 stdout_parts.append(result.output or "") 65 else: 66 error_parts.append(result.output or "No details provided") 67 68 if error_parts: 69 return ValidationResult( 70 status=ValidationStatus.ERROR, 71 error_message="\n".join(error_parts), 72 executed_code=executed, 73 ) 74 return ValidationResult( 75 status=ValidationStatus.SUCCESS, 76 stdout="\n".join(stdout_parts), 77 executed_code=executed, 78 )
- Imports (lines 1-4): Pull in the
google.genaiclient, thetypesmodule for tool declarations, Pydantic'sBaseModel/Fieldfor structured output, andEnumfor the validation status states. ValidationStatusenum (lines 6-9): Three outcomes —SUCCESSwhen code runs cleanly,ERRORwhen execution raises an exception, andTIMEOUTwhen the API call itself fails or exceeds limits.ValidationResultmodel (lines 11-15): Captures everything downstream consumers need: the status, any captured stdout, an error message if execution failed, and the exact code that was executed (useful for audit logging).__init__(lines 18-24): Initializes agenai.Client(which readsGOOGLE_API_KEYor application default credentials from the environment), stores the model identifier, sets the timeout, and pre-builds theToolCodeExecutiontool declaration so it is reused across calls.validateprompt construction (lines 27-31): Deliberately explicit — tells Gemini to execute the code without modification. Without this guard, Gemini might "improve" the code or add error handling, defeating the purpose of validating the original snippet.validateAPI call (lines 32-38): Passescode_execution_toolin the config and setstemperatureto 0.0 for deterministic execution behavior. The response object contains structured parts that_parse_responseinterprets.validateerror guard (lines 39-44): The outertry/exceptcatches network errors, quota exhaustion, and SDK failures, wrapping them in aTIMEOUTstatus so callers always receive a validValidationResultinstead of an unhandled exception._parse_responseiteration (lines 52-62): Walks the parts in the first candidate's content. When anexecutable_codepart is found, the.codeattribute (with anor executedfallback for None) records what Gemini actually ran. When acode_execution_resultpart is found, theoutcomeenum routes the output:OUTCOME_OKintostdout_parts, anything else (includingOUTCOME_DEADLINE_EXCEEDEDandOUTCOME_FAILED) intoerror_parts._parse_responseverdict (lines 64-74): If any error parts were collected, the status isERRORregardless of whether stdout was also captured. This conservative rule ensures that partially-failing code (e.g., prints output before raising an exception) is still flagged invalid.
Integrating Validation Into the Pipeline
With the CodeValidator built, integration into the broader message processing pipeline follows a straightforward pattern. After Instructor extracts CodeBlock models from the LLM response and the markdown parser confirms syntax validity via ast.parse, each Python block passes through CodeValidator.validate. The validation result attaches to the block before it enters the SSE streaming assembly phase. The following snippet shows how the orchestration layer calls the validator and annotates the block, using the ValidationResult to set a validated flag and optionally attach execution output to the content model that gets serialized into the SSE event stream.
Code snippet python
1from processing.code_validator import CodeValidator, ValidationStatus 2 3def process_code_blocks(blocks: list, validator: CodeValidator) -> list: 4 results = [] 5 for block in blocks: 6 if block.language != "python": 7 block.validated = False 8 block.validation_note = "non-python: skipped" 9 results.append(block) 10 continue 11 12 vr = validator.validate(block.source) 13 block.executed_code = vr.executed_code 14 block.validated = vr.status == ValidationStatus.SUCCESS 15 block.validation_note = ( 16 vr.stdout if vr.status == ValidationStatus.SUCCESS 17 else f"FAILED: {vr.error_message}" 18 ) 19 results.append(block) 20 return results
- Lines 1-2: Import the
CodeValidatorclass and theValidationStatusenum from the module built earlier in this lesson. - Lines 5-6: Initialize an empty results list and begin iterating over all code blocks extracted by the Instructor phase.
- Lines 7-11: Non-Python blocks (JavaScript, SQL, shell, etc.) skip validation entirely. The
validatedflag is set to False — not because the code is invalid, but because this validator only handles Python. Thevalidation_notefield communicates this to downstream consumers and the SSE client. - Lines 13-14: For Python blocks, call
validator.validatewith the raw source code. The returnedValidationResultcontains the status, stdout, error message, and the actual code that was executed in the sandbox. - Lines 15-19: Set the
validatedflag based on whether the status equalsValidationStatus.SUCCESS. Thevalidation_notefield carries either the captured stdout (useful for showing execution output in the UI) or a prefixed error message. This dual-purpose field avoids adding separatestdoutanderrorattributes to the block model, keeping the SSE payload compact. - Lines 20-21: Append the annotated block and return the complete list. The caller — typically the streaming assembly function from another goal — iterates this list to emit each block as a typed SSE event.
Do's and Don'ts
Do's
- ✓Do configure
ToolCodeExecutionas atypes.Toolobject and pass it through thetoolsparameter insideGenerateContentConfig— omitting either thetypes.Toolwrapper or theconfigargument causes Gemini to ignore the tool entirely and return a plain text response instead of executing the code, so the_parse_responseloop never finds acode_execution_resultpart. - ✓Do iterate over
response.candidates[0].content.partsand check both theexecutable_codeandcode_execution_resultattributes on each part — Gemini interleaves these part types in a single response, and skipping theexecutable_codecheck meansexecuted_codesilently falls back to the original string, hiding any pre-execution normalization Gemini applied before theOUTCOME_OKorOUTCOME_ERRORverdict. - ✓Do keep
CodeValidatorstateless — only a sharedgoogle.genai.Clientand model identifier, no per-call execution state — statelessness is what makes it safe to fan out concurrent validations withasyncio.gatherwhen a single LLM response contains multiple Python blocks, bounding SSE emission latency to the slowest block rather than the cumulative sum.
Don'ts
- ✗Don't import from the
google-generativeaipackage; usefrom google import genai(thegoogle-genaiSDK) —types.ToolCodeExecution,types.Outcome, andtypes.GenerateContentConfiglive only in the newer SDK's namespace; the older package lacks these types entirely, causingAttributeErrorat thegenerate_contentcall site with no fallback behavior. - ✗Don't set
temperatureabove0.0inGenerateContentConfigwhen invoking the code execution tool — any non-zero temperature gives Gemini latitude to rewrite the snippet before executing it, so anOUTCOME_OKverdict reflects a modified version of the code rather than the original extracted block, breaking the pipeline's guarantee that only the emitted-as-extracted code has been proven runnable. - ✗Don't omit the
timeoutparameter or leave exception handling to the caller — a Python block with an infinite loop or blocking I/O will stallCodeValidator.validateindefinitely, freezing the synchronous SSE gate and preventing any downstream content from reaching the client; the 30-second default and the broadexcept Exceptioncatch that maps toValidationStatus.TIMEOUTare what make the validator safe to insert as a blocking pipeline stage.
This lesson is free to read. Its 3 hands-on labs — real code, in a cloud IDE — are part of the GenAI Application Engineering subscription.
From · cancel anytime · Already a subscriber? Sign in →
More free lessons in Full-Stack GenAI Applications
- Ch 2Extract structured output with Instructor + Pydantic
- Ch 2Build a usage logging system with token + cost capture
- Ch 3Build a context window composer with token budgets
- Ch 3Implement Anthropic prompt caching with cache_control markers
- Ch 4Build a code validator with Gemini ToolCodeExecutionYou are here
- Ch 5Build a Pydantic AI regeneration agent with typed tools
- Ch 6Build a document chunking pipeline (recursive, semantic, token-aware)