new·The score now tells you which way it movedA brain's exam only ever grows: its own material writes questions, and so does every question a real caller asked and did not get answered. The score is a percentage over that growing set, so a brain that learned more could post a smaller number — and this week three did. One of them answered two MORE questions than the week before and showed eighteen points less. Printed as a single percentage, that reads as decline to a reader and as punishment to anyone who contributes material.all news →
mozg.beta
Sign in

Temporal · Develop · all subjects

integrations

245 notes in this subject, read out of this brain and free to use. This is page 5 of 5.

File upload in Workflow example

@workflow.defn class FilesWorkflow: @workflow.run async def run(self, file_path: str, prompt: str) -> str: client = TemporalAsyncClient(); uploaded = await client.files.upload(file=file_path, config=types.UploadFileConfig(mime_type="text/plain")); contents = cast(types.ContentListUnion, [prompt, uploaded]); response = await client.models.generate_content(model="gemini-2.5-flash", contents=contents); return response.text or ""

Vertex AI Workflow example

@workflow.defn class VertexAIWorkflow: @workflow.run async def run(self, prompt: str, project: str, location: str) -> str: client = TemporalAsyncClient(vertexai=True, project=project, location=location); response = await client.models.generate_content(model="gemini-2.5-flash", contents=prompt); return response.text or ""

Vertex AI Worker setup example

genai_client = genai.Client(vertexai=True, project=os.environ["GOOGLE_CLOUD_PROJECT"], location=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")); plugin = GoogleGenAIPlugin(genai_client)

Timeout and retry override for all API calls

from datetime import timedelta; from temporalio.common import RetryPolicy; from temporalio.contrib.google_genai import TemporalAsyncClient; from temporalio.workflow import ActivityConfig; client = TemporalAsyncClient(activity_config=ActivityConfig(start_to_close_timeout=timedelta(minutes=5), retry_policy=RetryPolicy(maximum_attempts=3)))

Interactions API Workflow example

@workflow.defn class InteractionsWorkflow: @workflow.run async def run(self, prompt: str) -> dict[str, Any]: client = TemporalAsyncClient(); interaction: Any = await client.interactions.create(model="gemini-2.5-flash", input=prompt); fetched: Any = await client.interactions.get(interaction.id); await client.interactions.delete(interaction.id); return {"id": interaction.id, "status": str(fetched.status)}

Give your agent this brain