cv

Author: networkandcode

gemma4-try/agent.py

@@ -11,8 +11,4 @@
     MODEL,
 )
 
-search_agent = Agent(
-    name=AGENT_NAME,
-    model=MODEL,
-    tools=[fetch_fruits, greet_user]
-)
+search_agent = Agent(name=AGENT_NAME, model=MODEL, tools=[fetch_fruits, greet_user])

gemma4-try/main.py

@@ -9,27 +9,28 @@
     SESSION_ID,
     USER_ID,
 )
-    
+
+
 async def main():
     await session_service.create_session(
-        app_name=APP_NAME,
-        user_id=USER_ID,
-        session_id=SESSION_ID
+        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
     )
-    
+
     while True:
         user_message = input("User: ")
-        content = types.Content(role='user', parts=[types.Part(text=user_message)])
-        
-        async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
+        content = types.Content(role="user", parts=[types.Part(text=user_message)])
+
+        async for event in runner.run_async(
+            user_id=USER_ID, session_id=SESSION_ID, new_message=content
+        ):
             if event.is_final_response():
                 if event.content and event.content.parts:
                     for part in event.content.parts:
-                        if (not part.thought):
+                        if not part.thought:
                             print(part.text)
             else:
                 print(event)
-                
+
 
 if __name__ == "__main__":
     asyncio.run(main())

gemma4-try/runner.py

@@ -3,11 +3,7 @@
 from agent import search_agent
 from session_service import session_service
 from vars import (
-    APP_NAME, 
+    APP_NAME,
 )
 
-runner = Runner(
-    agent=search_agent,
-    app_name=APP_NAME,
-    session_service=session_service
-)
+runner = Runner(agent=search_agent, app_name=APP_NAME, session_service=session_service)

gemma4-try/tool.py

@@ -1,5 +1,6 @@
 import httpx
 
+
 def fetch_fruits(fruit: str):
     url = f"https://api.example.com/api/fruit/{fruit}"
 
@@ -12,4 +13,4 @@ def fetch_fruits(fruit: str):
 
 
 def greet_user(user: str):
-    return f"Hello {user}"
+    return f"Hello {user}"

gemma4-try/vars.py

@@ -1,5 +1,5 @@
-AGENT_NAME="test_agent"
+AGENT_NAME = "test_agent"
 APP_NAME = "test_app"
-MODEL="gemma-4-31b-it" # gemma-4-26b-a4b-it is also supported
+MODEL = "gemma-4-31b-it"  # gemma-4-26b-a4b-it is also supported
 SESSION_ID = "test_session"
 USER_ID = "test_user"

src/data/cv.json

@@ -2,24 +2,25 @@
 
 from strands.agent.a2a_agent import A2AAgent
 
+
 async def main():
     agent = A2AAgent(endpoint="http://localhost:9000")
     agent_card = await agent.get_agent_card()
-    
+
     print("Invoking remote agent with agent card:")
     for key, value in agent_card:
         print(key, ":", value)
-    print('-' * 20)
-    
+    print("-" * 20)
+
     while True:
         prompt = input("User: ")
         if prompt.lower() == "exit":
             break
-            
+
         async for event in agent.stream_async(prompt):
             if "result" in event:
                 print("AI: ", event["result"], end="", flush=True)
 
-            
+
 if __name__ == "__main__":
     asyncio.run(main())

strands-a2a-demo/strands-a2a-client/main.py

@@ -5,6 +5,7 @@
 
 load_dotenv()
 
+
 def main():
     agent = Agent(
         callback_handler=None,
@@ -16,5 +17,6 @@ def main():
     a2a_server = A2AServer(agent=agent)
     a2a_server.serve()
 
+
 if __name__ == "__main__":
     main()

strands-a2a-demo/strands-a2a-server/main.py

@@ -7,6 +7,7 @@
 
 load_dotenv()
 
+
 async def main():
 
     agent = Agent(
@@ -16,10 +17,10 @@ async def main():
 
     st.title("Strands App")
 
-    if 'messages' not in st.session_state:
-        st.session_state['messages'] = []
+    if "messages" not in st.session_state:
+        st.session_state["messages"] = []
 
-    for message in st.session_state['messages']:
+    for message in st.session_state["messages"]:
         role = message["role"]
         content = message["content"]
 
@@ -42,8 +43,8 @@ async def main():
             stream = agent.stream_async(prompt)
             placeholder = st.empty()
             async for chunk in stream:
-                if 'data' in chunk:
-                    result += chunk['data']
+                if "data" in chunk:
+                    result += chunk["data"]
                     placeholder.markdown(result)
                     time.sleep(0.05)
 
@@ -53,6 +54,7 @@ async def main():
                     "content": result,
                 }
             )
-        st.session_state['messages'] += messages
+        st.session_state["messages"] += messages
+
 
 asyncio.run(main())

strands-examples/app.py

@@ -6,16 +6,18 @@
 from strands.tools.mcp import MCPClient
 import streamlit as st
 
+
 def remove_html_tags(text_with_html):
     text_with_out_html = re.sub(r"<[^>]+>", "", text_with_html)
     return text_with_out_html
 
+
 async def stream_result(stream):
     result = ""
     placeholder = st.empty()
     async for chunk in stream:
-        if 'data' in chunk:
-            result += chunk['data']
+        if "data" in chunk:
+            result += chunk["data"]
             result = remove_html_tags(result)
             placeholder.write(result)
 
@@ -26,33 +28,23 @@ def __init__(self):
 
         server_params = {
             "command": "npx",
-            "args": [
-                "-y",
-                "kubernetes-mcp-server@latest"
-            ],
-            "env": {
-                "KUBECONFIG": "k3s.yaml"
-            }
+            "args": ["-y", "kubernetes-mcp-server@latest"],
+            "env": {"KUBECONFIG": "k3s.yaml"},
         }
-        
-        self.stdio_mcp_client = MCPClient(lambda: stdio_client(
-            StdioServerParameters(
-                **server_params
-            )
-        ))
+
+        self.stdio_mcp_client = MCPClient(
+            lambda: stdio_client(StdioServerParameters(**server_params))
+        )
 
         with self.stdio_mcp_client:
             # Get the tools from the MCP server
             tools = self.stdio_mcp_client.list_tools_sync()
 
             # Create an agent with these tools
             self.agent = Agent(
-                callback_handler=None,
-                model="us.amazon.nova-micro-v1:0",
-                tools=tools
+                callback_handler=None, model="us.amazon.nova-micro-v1:0", tools=tools
             )
 
-    
     async def send_prompt(self, prompt):
         with self.stdio_mcp_client:
             stream = self.agent.stream_async(prompt)