mitzen

We are going to look at a simple kubeflow condition and parallel task when working with pipeline, This pipeline will greet pet owner and if there's pets we greet them too.  This is our example components:- @component def sayhi ( name : str ) -> int :   return 1 if name == "eric" else 2 @component def args_generator_op () -> list :     return [{ "cats" : "1" , "dogs" : "2" }, { "cats" : "10" , "dogs" : "20" }] And this is our pipeline code:- import json from kfp import dsl @dsl.pipeline (name= "control" , pipeline_root=PIPELINE_ROOT) def pipeline ():     name = sayhi(name= "greet-human" )       with dsl.If(name.output == 1 , "if has pets" ):           cat_and_dog = args_generator_op()           with dsl.Else( "if has pets" ):         cat_and_dog = args_generator_op()       with dsl.ParallelFor(cat_and_dog.output, name= "hi-to-pe...

We can easily run ML pipeline using aiplatfor sdk using the following setup - notebook  PROJECT_ID = "project-your-id"   # @param {type:"string"} LOCATION = "us-central1"   # @param {type:"string"} BUCKET_URI = f "gs://your-bucket-name- {PROJECT_ID} -unique"   # @param {type:"string"} ! gcloud storage buckets create --location={LOCATION} --project={PROJECT_ID} {BUCKET_URI} SERVICE_ACCOUNT = "ml-pipeline-dev-sa"   # @param {type:"string"} import sys IS_COLAB = "google.colab" in sys.modules if (     SERVICE_ACCOUNT == ""     or SERVICE_ACCOUNT is None     or SERVICE_ACCOUNT == "ml-pipeline-dev-sa" ):     # Get your service account from gcloud     if not IS_COLAB:         shell_output = ! gcloud auth list 2 >/dev/null         SERVICE_ACCOUNT = shell_output[ 2 ].replace( "*" , "" ).strip()     if IS_COLAB:         shell_output = !...

We can create gcp agent platform pipeline using kubeflow or tensorflow SDK (this is for large dataset which span terrabytes).  Kubeflow sdk are basically python script that you defines pipelines or components. @kfp.dsl.pipeline and @kfp.dsl.componet. Lets gets started by installing the required modules  uv install kfp.dsl  google-cloud-pipeline-components  docker In its simplistic form,  a pipeline can look like this (this will not work for Windows setup with docker) local_pipeline.py from kfp import dsl # Define a component to add two numbers. @ dsl . component def add ( a : int , b : int ) -> int :     return a + b # Define a simple pipeline using the component. @ dsl . pipeline def addition_pipeline ( x : int , y : int , z : int ) -> int :     task1 = add ( a = x , b = y )     task2 = add ( a = task1 . output , b = z )     return task2 . output And here is our runner  from kfp import local fro...

Let's say you would like to ensure that you pipeline operates in a sequential manner - lockBehaviour  option- sequential in a sense that if user A trigger pipeline at 12:00 and then another user B trigger pipeline at 12:01, user B will have to wait until user A completes.  If user a cancel it, User B will have to wait until the cancellation completes. This can take a while before next pipeline start executing.  There are 2 option available - sequential and runLatest. stages: - stage: A   lockBehavior: sequential   jobs:   - job: Job     steps:     - script: Hey! If we do not specify, lockBehaviour is runLatest. Potential issue :- Queued runs waiting on a check can eventually time out, so B could fail if it waits too long behind other runs. Only runs targeting the same protected resource (for example, the same environment, repository, agent pool, service connection) compete for the lock.

With Azure Devops future uncertain - find myself hardly read or keep up to date with it. Well just in case, I do, there is always this link here to goto  :) https://learn.microsoft.com/en-us/azure/devops/release-notes/2026/sprint-280-update

In gcp agentic platform mcp server is an abtraction that allows your agent to call a actual MCP hosted elsewhere.  It is not to create and deploy an MCP server. You can of course do that via Google Cloud run.  Typically you can use the following code in your agentic runtime to call this MCP server  import os from google.adk.integrations.agent_registry import AgentRegistry from google.auth import default from google.adk.agents import Agent from google.adk.models import Gemini from google.genai import types _ , project_id = default() LOCATION = os.environ.setdefault( "GOOGLE_CLOUD_LOCATION" , "us-central1" ) MCP_SERVER_NAME = os.environ.get( "MCP_SERVER_NAME" , "agentregistry-00000000-0000-0000-bd07-35253020786b" ) os.environ[ "GOOGLE_GENAI_USE_VERTEXAI" ] = "True" registry = AgentRegistry(project_id=project_id, location=LOCATION) mcp_toolset = registry.get_mcp_toolset(     f "projects/project-id/locations/us-cen...

 This is an example of how we can deployment agent runtime agent with environment variables. The key settings here is the "env_vars".  import asyncio from urllib import response from google . adk . agents import Agent from google . adk . runners import Runner from google . adk . sessions import InMemorySessionService from google . genai import types from google . adk . agents . callback_context import CallbackContext from google . adk . models . llm_request import LlmRequest from vertexai import types as vertex_types from google . genai import types import vertexai import agentplatform from agentplatform import types   # check below: this is my assumption from agentplatform . frameworks import AdkApp # 1. Define the callback function def inspect_llm_input ( callback_context : CallbackContext , llm_request : LlmRequest ):     """Callback triggered directly before ADK sends payload to the LLM."""     print ( ...

When you're calling your agentic app, you ran into this error  raise ClientError(status_code, response_json, response) google.genai.errors.ClientError: 404 NOT_FOUND. {'error': {'code': 404, 'message': 'Not Found. However, the operation was successful.', 'status': 'NOT_FOUND'}} And then you try to see what is wrong. This can lead into a whole rabbit whole. Took me 2 hours to resolve this.  This happens to me because I enabled gateway agent on my agent. So an operation that will generally be fine, has raise this error.  I had to disabled that be redeploy my runtime and removing gateway from it.

So what happens after you have deployed your agent to agentic platform, you should have code to call it. How can we do that? import asyncio import agentplatform client = agentplatform .Client(     project = "project-your-project-id" ,     location = "us-central1" , ) async def main ():     runtime = client .runtimes.get(         name = "projects/your-project-id/locations/us-central1/ reasoningEngines/2026277059566764032"     )     print ( runtime )     runtime .operation_schemas()     session = await runtime .async_create_session( user_id = "user_123" )     async for event in runtime .async_stream_query(         user_id = "user_123" ,         message = "What is my account balance including pending charges for user_123?" ,     ):         print ( event ) asyncio . run ( main ()) It is even documente...